Clarify validator semantics and downstream dedupe expectations

[shu1513]

Summary by CodeRabbit

• New Features

  • Added a state-resources writer pipeline and a runnable script to persist validated state resource items to the database.
  • Exposed a configurable run mode (one-shot or continuous) for the writer.

• Chores

  • Added new pipeline identifiers and improved validation, error handling, reclaim/retry recovery, and completion metrics.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 00f74add-b3cc-494d-84b2-0f5da132a175

📥 Commits

Reviewing files that changed from the base of the PR and between f2ed9d0 and 2745de7.

📒 Files selected for processing (2)

• backend/src/pipeline/validators/stateResourcesValidator.ts
• backend/src/pipeline/writers/stateResourcesWriter.ts

✅ Files skipped from review due to trivial changes (1)

• backend/src/pipeline/writers/stateResourcesWriter.ts

🚧 Files skipped from review as they are similar to previous changes (1)

• backend/src/pipeline/validators/stateResourcesValidator.ts

---

📝 Walkthrough

Walkthrough

Adds a new state_resources writer process and script, introduces staging/written stream and writer consumer-group constants, and refactors validator logic to reclaim and centrally handle pending Redis stream entries with updated DB status handling and error semantics.

Changes

Cohort / File(s)	Summary
Package & Script backend/package.json, backend/src/scripts/writeStateResources.ts	Added state-resources:write npm script and a CLI script that runs the writer with optional --once, logging failures and exiting on error.
Pipeline Config backend/src/config/stateResourcePipeline.ts	Added exported constants STAGING_WRITTEN_STREAM and STAGING_STATE_RESOURCES_WRITER_GROUP.
Validator Logic backend/src/pipeline/validators/stateResourcesValidator.ts	Refactored consumption to reclaim stale pending entries (XAUTOCLAIM), unified entry handling via handleEntries(), added retried accounting, safer DB failure-marking (only when row still pending), and clarified at-least-once semantics with comments.
Writer Implementation backend/src/pipeline/writers/stateResourcesWriter.ts	New exported runStateResourcesWriter(options) that consumes validated stream entries, validates payloads, performs transactional upserts to state_resources, transitions staging → written, publishes to staging:written, and manages Redis consumer-group lifecycle and error/recovery paths.

Sequence Diagram

sequenceDiagram
    participant Redis as Redis Stream\n(validated)
    participant Writer as StateResourcesWriter
    participant Staging as Postgres Staging Table
    participant DB as Postgres DB
    participant StgStream as Redis Stream\n(staging:written)

    loop Batch loop
        Writer->>Redis: XREADGROUP (fetch batch / pending reclaim)
        Redis-->>Writer: messages (ids + payloads)
        alt Per message
            Writer->>Writer: Parse & validate payload
            alt Invalid payload
                Writer->>Staging: UPDATE status → failed (with reason)
                Writer->>Redis: XACK id
            else Valid payload
                Writer->>Staging: SELECT staging row by ingest_key
                alt row missing or not validated
                    Writer->>Redis: XACK id (skipped)
                else row validated
                    Writer->>DB: BEGIN
                    Writer->>DB: UPSERT state_resources (by state_fips)
                    Writer->>Staging: UPDATE status → written
                    Writer->>DB: COMMIT
                    Writer->>StgStream: XADD staging:written ...
                    Writer->>Redis: XACK id
                end
            end
        end
    end
    Writer->>Redis: QUIT
    Writer->>DB: Close pool

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~50 minutes

Possibly related PRs

• Add state_resources validator and state_resources text limits #2 — touches the same validator logic for state_resources and may overlap with the validator refactor and text/limit changes.

Poem

🐇 I hopped through streams at break of day,

Validated, written, then bounced away,
Transactions snug, and staging sighed,
Redis thanked me for what I tried,
A little rabbit, keeping data gay.

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Title check	⚠️ Warning	The PR title does not accurately reflect the primary changes, which involve implementing a new state resources writer pipeline with Redis/Postgres integration; the title focuses only on validator semantics clarification.	Revise the title to reflect the main implementation work, such as 'Implement state resources writer pipeline with validator recovery semantics' or similar, that captures the new writer module and recovery mechanisms.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch codex/state-resources-writer

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 3

🧹 Nitpick comments (1)

backend/src/pipeline/writers/stateResourcesWriter.ts (1)

61-121: Consider reusing the validator's payload schema here.

This parser is already looser than backend/src/pipeline/validators/stateResourcesValidator.ts: it no longer checks HTTP(S) URLs, extra sources keys, or the FIPS/abbreviation mapping. Pulling normalization/validation into a shared helper would keep validator and writer semantics from drifting.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@backend/src/pipeline/writers/stateResourcesWriter.ts` around lines 61 - 121,
The parseStateResourcePayload implementation diverges from the stricter
validator (stateResourcesValidator.ts); update it to reuse the validator's
payload schema/normalizer instead of reimplementing checks: import and call the
shared validation/normalization function used by
backend/src/pipeline/validators/stateResourcesValidator.ts (or extract the
validator logic into a new helper like normalizeStateResourcePayload), then in
parseStateResourcePayload delegate all checks (HTTP(S) URL format, allowed
sources keys, FIPS/abbreviation mapping, citation shape) to that helper and
return its normalized payload on success or its error reason on failure;
reference parseStateResourcePayload, StateResourcePayload, StateResourceSources
and the validator module when locating where to plug in the shared helper.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@backend/src/pipeline/validators/stateResourcesValidator.ts`:
- Around line 318-320: The current behavior moves rows to "failed" when
redis.xAck() throws after a successful redis.xAdd(STAGING_VALIDATED_STREAM, ...)
and validated DB update, which causes the published validated event to be
skipped by stateResourcesWriter; change this so post-publish ack failures do NOT
transition the DB row to "failed". Specifically, in the code path that calls
redis.xAdd(...) and then redis.xAck(...), catch ack errors separately: do not
perform the existing failed-state rewrite (the logic in the outer catch handling
rows -> "failed" and acking the pending message); instead log the ack error with
context (ingest_key, stream entry id), schedule/attempt a retry for xAck (or
push the entry id to a dedicated retry queue), and leave the DB row in
"validated" so downstream dedupe/consumers can still process the published
event; update handling around STAGING_VALIDATED_STREAM, redis.xAdd, and
redis.xAck to reflect this separation of post-publish ack failures from state
transitions.

In `@backend/src/pipeline/writers/stateResourcesWriter.ts`:
- Around line 328-334: The worker currently only calls redis.xReadGroup(...)
with id: ">" and never reclaims pending messages, so add an XAUTOCLAIM pass
before entering the while (keepRunning) loop to claim and reprocess any pending
entries for the STAGING_STATE_RESOURCES_WRITER_GROUP consumer group for this
consumerName on STAGING_VALIDATED_STREAM; call redis.xAutoClaim (or XAUTOCLAIM
equivalent) to fetch pending IDs (e.g., min-idle-time 0 or configured threshold)
and re-enqueue/process them the same way you handle xReadGroup results, ensuring
xAck still occurs after successful processing and referencing the existing
xReadGroup/xAck flow to integrate claims.

---

Nitpick comments:
In `@backend/src/pipeline/writers/stateResourcesWriter.ts`:
- Around line 61-121: The parseStateResourcePayload implementation diverges from
the stricter validator (stateResourcesValidator.ts); update it to reuse the
validator's payload schema/normalizer instead of reimplementing checks: import
and call the shared validation/normalization function used by
backend/src/pipeline/validators/stateResourcesValidator.ts (or extract the
validator logic into a new helper like normalizeStateResourcePayload), then in
parseStateResourcePayload delegate all checks (HTTP(S) URL format, allowed
sources keys, FIPS/abbreviation mapping, citation shape) to that helper and
return its normalized payload on success or its error reason on failure;
reference parseStateResourcePayload, StateResourcePayload, StateResourceSources
and the validator module when locating where to plug in the shared helper.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 18b87888-e077-4a1c-bc93-991bc96334d6

📥 Commits

Reviewing files that changed from the base of the PR and between 2aa767e and f2ed9d0.

📒 Files selected for processing (5)

• backend/package.json
• backend/src/config/stateResourcePipeline.ts
• backend/src/pipeline/validators/stateResourcesValidator.ts
• backend/src/pipeline/writers/stateResourcesWriter.ts
• backend/src/scripts/writeStateResources.ts