​

Phase 0B — Red-Flag & Alignment Scan

​

1. BigQuery Schema Confirmation

​

1.1 error_codes & warnings Types

• error_codes: STRING, REPEATED (ARRAY<STRING>)
• warnings: STRING, REPEATED (ARRAY<STRING>)
• No nullable mode (REPEATED arrays are never NULL, empty array = [])

​

1.2 updated_at / updated_by Existence

CURRENT_TIMESTAMP() AS updated_at,
@created_by AS updated_by
-- Comment: "Optional: updated_at/updated_by would be set here if columns exist"

• ❌ updated_at: NOT PRESENT in table schema
• ❌ updated_by: NOT PRESENT in table schema

​

1.3 Required/Not-Null Fields (Staging Must Always Populate)

1. tenant_id (STRING)
2. batch_id (STRING)
3. row_index (INTEGER)
4. event_type (STRING)
5. period_label (STRING)
6. business_name_raw (STRING)
7. mapped_payload_json (STRING)
8. row_status (STRING)
9. created_at (TIMESTAMP)
10. created_by (STRING)

• Must populate all REQUIRED fields
• Optional fields can be NULL (source_system, member_id_raw, payee_agent_id_raw, credit_raw, debit_raw, total_raw, period_code_raw, pay_periods_raw)
• Arrays (error_codes, warnings) can be empty [] but not NULL (REPEATED mode)

​

2. Load Job Feasibility

​

2.1 trigger_load_job Support Analysis

• Format: bigquery.SourceFormat.CSV only
• Schema: autodetect=True (no explicit schema)
• Write Disposition: WRITE_TRUNCATE only

• ✅ NDJSON Support: bigquery.SourceFormat.NEWLINE_DELIMITED_JSON (supported by BigQuery)
• ✅ Explicit Schema: schema=[...] parameter (supported, must provide)
• ✅ WRITE_APPEND: bigquery.WriteDisposition.WRITE_APPEND (supported)

​

2.2 Sync /map Timing Estimate (Dec File: 2,841 rows)

1. Parse + process_row(): ~5-10s (in-memory, Python)
2. Write NDJSON to GCS: ~2-5s (821 KB file)
3. Trigger Load Job: ~1s (async job creation, returns immediately)
4. Poll Load Job: ~10-30s (BigQuery load job for 2,841 rows NDJSON)
5. MERGE Staging → Target: ~5-10s (single MERGE statement)
6. Verify Counts: ~2-3s (2 COUNT queries)
7. Finalize Status: ~1s (UPDATE batch status)

​

2.3 Async Fallback Threshold

• 10k rows NDJSON ≈ ~3MB file
• Load job timing: ~30-60s (still under 300s, but approaching limit)
• Provides safety margin for network latency, BigQuery queue delays
• Allows graceful degradation for very large files

• Check len(processed_rows) > ASYNC_THRESHOLD (default 10,000)
• If async: Return 202 with load_job_id, status="PROCESSING"
• Frontend polls /api/v1/intake/{batch_id}/status until status="MAPPED"

​

3. Concurrency/Idempotency

​

3.1 Call Paths to /map

1. Frontend UI (dashboard/src/lib/intakeClient.ts):
   • User clicks “Map Columns” button
   • POST /api/v1/intake/{batch_id}/map
   • Single user action, no concurrent calls expected
2. API Route (api/routes/intake.py:316):
   • @router.post("/{batch_id}/map")
   • RBAC: require_admin_or_ceo
   • No background jobs or async tasks call this endpoint

​

3.2 MAPPING_IN_PROGRESS Lock Strategy

• UPLOADED → MAPPED (current)
• MAPPED → MAPPED (re-materialize if mapping changed)

• User clicks “Map” twice rapidly → Both requests fetch status=UPLOADED
• Both proceed to materialize rows → Duplicate writes (idempotent via MERGE, but wasteful)
• Both attempt UPDATE status=MAPPED → Last write wins (no conflict)

# Step 1: Atomic status check + transition
UPDATE ingestion_batches
SET status = 'MAPPING_IN_PROGRESS'
WHERE batch_id = @batch_id
  AND tenant_id = @tenant_id
  AND status = 'UPLOADED'  -- Only transition from UPLOADED

• Status is not UPLOADED → Return 409 CONFLICT (“Mapping already in progress or completed”)

• Lock acquired → Proceed with materialization
• On success: UPDATE status='MAPPED'
• On failure: UPDATE status='UPLOADED' (rollback) OR UPDATE status='ERROR'

​

3.3 Idempotent Shortcut (Stale Status Handling)

• Checks status == "MAPPED" AND verified_count > 0 AND mapping_hash matches
• Returns 200 immediately (short-circuit)

• Request 1: Materializes rows, writes to BigQuery, but UPDATE status='MAPPED' fails
• Request 2: Fetches batch → status=UPLOADED (stale)
• Request 2: Checks verified_count → Finds rows exist → Compares mapping_hash → Matches → Returns 200 ✅

​

4. Audit Semantics

​

4.1 created_at / created_by Immutability

-- Audit-grade: created_at and created_by are NOT updated on MATCHED (immutable)
-- Optional: updated_at/updated_by would be set here if columns exist

• Must preserve immutability: created_at and created_by from staging (source) are used for INSERT
• On MATCHED: Do NOT update created_at/created_by (only update data fields)

​

4.2 Lineage Retention (NDJSON Deletion)

1. ✅ trace_id: Already logged and returned in response
2. ✅ load_job_id: Must be logged and optionally stored in ingestion_batches metadata
3. ✅ merge_job_id: Must be logged (from MERGE query job)
4. ⚠️ gcs_uri: Optional - Can be stored in ingestion_batches metadata field (if added) or logged only

• Store in ingestion_batches: Add optional metadata JSON field (or reuse existing fields)
• Log: All job IDs + trace_id + gcs_uri (for Cloud Logging searchability)
• TTL: Delete NDJSON after 7 days (or after MERGE succeeds + 1 day grace period)

ALTER TABLE ingestion_batches ADD COLUMN IF NOT EXISTS metadata STRING;
-- JSON: {"load_job_id": "...", "merge_job_id": "...", "staging_gcs_uri": "...", "staged_count": 2841, "target_count": 2841}

​

5. Failure-Mode Matrix

​

5.1 Failure Points & Error Contracts

{
  "error": "<StableErrorCode>",
  "reason": "<safe_message>",
  "trace_id": "<trace_id>",
  "tenant_id": "<tenant_id>",
  "batch_id": "<batch_id>",
  "period_label": "<period_label>",
  "route": "/api/v1/intake/{batch_id}/map",
  "load_job_id": "<optional>",
  "merge_job_id": "<optional>",
  "expected_count": "<optional>",
  "staged_count": "<optional>",
  "target_count": "<optional>"
}

​

6. Cleanup Safety

​

6.1 TTL vs Delete-by-Batch

• Pattern: GCS lifecycle policy (delete objects older than 7 days)
• Pros: Automatic, no code complexity
• Cons: Cannot delete immediately after success, may retain failed staging data

• Pattern: After MERGE succeeds + count verification, delete gs://bucket/intake-staging/{tenant_id}/{batch_id}/*
• Pros: Immediate cleanup, only deletes successful batches
• Cons: Requires explicit cleanup code, must handle failures gracefully

• Immediate: Delete staging NDJSON after MERGE succeeds (best-effort, non-blocking)
• Fallback: GCS lifecycle policy (delete objects older than 7 days) for orphaned staging files

​

6.2 Cleanup Isolation (Cannot Delete Other Batches)

gs://{bucket}/intake-staging/{tenant_id}/{batch_id}/transaction_events_raw.ndjson

staging_path = f"intake-staging/{tenant_id}/{batch_id}/transaction_events_raw.ndjson"
blob = bucket.blob(staging_path)
blob.delete()  # Only deletes this specific batch_id's file

• ✅ Tenant isolation: Path includes tenant_id (from JWT, not user input)
• ✅ Batch isolation: Path includes batch_id (from route parameter, validated UUID)
• ✅ No wildcards: Delete specific blob, not prefix-based delete
• ✅ Best-effort: Wrap in try/except, log but don’t fail if delete fails

​

7. Red Flags Identified

​

🔴 Critical (Must Fix Before Implementation)

1. No MAPPING_IN_PROGRESS Lock: Race condition possible if user clicks “Map” twice rapidly
   • Fix: Add atomic status transition check
2. mapping_hash Not Stored: Idempotency relies on computing hash on-the-fly
   • Fix: Store mapping_hash in ingestion_batches during update_batch_mapping()
3. updated_at/updated_by Referenced But Don’t Exist: MERGE statement references non-existent columns
   • Fix: Remove from staging MERGE statement (safe, but cleanup needed)

​

🟡 Medium (Should Fix)

4. No Async Threshold: All requests are sync, even for very large files
   • Fix: Add async fallback for >10k rows
5. No Cleanup Strategy: Staging NDJSON files may accumulate
   • Fix: Implement hybrid cleanup (immediate + TTL)

​

🟢 Low (Nice to Have)

6. No Metadata Field: Cannot store load_job_id, merge_job_id, staging_gcs_uri
   • Fix: Add metadata JSON field to ingestion_batches (optional)

​

8. Implementation Readiness

​

✅ Ready to Proceed

• Schema confirmed (error_codes, warnings are REPEATED STRING)
• Load job supports NDJSON + explicit schema + WRITE_APPEND
• Timing estimate: 26-60s for 2,841 rows (well under 300s)
• Idempotency works even with stale status
• Audit semantics preserved (created_at/created_by immutable)

​

⚠️ Must Address Before Implementation

1. Add MAPPING_IN_PROGRESS status + atomic transition
2. Store mapping_hash in ingestion_batches
3. Remove updated_at/updated_by from staging MERGE (they don’t exist)
4. Implement structured error contracts for all failure points
5. Add cleanup logic (immediate delete + TTL fallback)