SQS queues

Sophia defines two primary queues and two DLQs in infra/queues.ts. SST resource names determine AWS queue names (e.g. sophia-dev-InterviewCompletedQueue-<suffix>); see comments in infra/queues.ts for naming rationale.

---

1. Interview completed (terminal interviews)

Queue name

• SST component: InterviewCompleted → interviewCompletedQueue
• DLQ component: InterviewCompletedDlq → interviewCompletedDlq
• Deploy output: interviewCompletedQueue, interviewCompletedDlq (sst.config.ts)

Producer

Trigger	Service	Function
Interview marked completed (bot callback)	Fargate API	completeInterview → publishInterviewCompleted
Interview marked failed (bot callback)	Fargate API	failInterview → publishInterviewFailed

Sources:

• apps/api/src/modules/interviews/bot-interview.service.ts (completeInterview, failInterview)
• apps/api/src/modules/interviews/sqs.service.ts (publishInterviewCompleted, publishInterviewFailed, internal publishToInterviewQueue)

Config: INTERVIEW_COMPLETED_QUEUE_URL (infra/api.ts, packages/shared/src/env.ts). If missing, publish is skipped with a warning.

Message attributes (all publishes to this queue):

Attribute	Type	Values
requestId	String	Current API request id from getRequestId() or 'unknown' (requestIdForPublish)
eventType	String	interview.completed or interview.failed

Message schema / format

JSON body (no envelope). Consumer validates with a Zod union in interview-completed.ts.

Completed interview (interviewCompletedSchema):

{
  "interviewId": "<ObjectId string>",
  "organizationId": "<ObjectId string>",
  "completedAt": "<ISO-8601 datetime>",
  "durationSeconds": 0
}

Failed interview (interviewFailedSchema in interview-failed.ts, matched when body includes type):

{
  "type": "interview.failed",
  "interviewId": "<ObjectId string>",
  "organizationId": "<ObjectId string>",
  "failureReason": "<string>",
  "failedAt": "<ISO-8601 datetime>"
}

Producer defaults: completedAt / failedAt default to new Date(); durationSeconds defaults to 0 if omitted (sqs.service.ts).

Consumer (Lambda)

Property	Value
Function name	`${$app.name}-${$app.stage}-interview-completed`
Handler	apps/functions/src/handlers/interview-completed.handler
Wrapper	createSqsConsumer (apps/functions/src/platform/sqs-consumer.ts)
Routing	If 'type' in payload → failed path; else completed path

Failed-only handler interview-failed.handler exists but is not subscribed in infra/queues.ts.

Processing logic

1. Connect Mongo (connectMongo in wrapper).
2. Parse and validate body; invalid → log and drop.
3. Failed: settleFailedInterview(interviewId, organizationId, failedAt).
4. Completed: settleCompletedInterview(interviewId, organizationId, completedAt, durationSeconds).

Implementation: apps/functions/src/handlers/interview-completed.ts → apps/api/src/modules/credits/settlement.service.ts (re-exported from apps/functions/src/handlers/settlement.ts).

Completed settlement:

• billedCredits = floor(durationSeconds / 60); cap consumed to reserved credits.
• Transaction: claim interview (settledAt: null), set credits.consumed, retentionExpiresAt (60 days from completion anchor).
• settleCreditsInSession for ledger/lot/org updates.
• If this invocation actually settled (settled === true), call publishReportGenerate with request id settlement-${interviewId} (errors logged, not thrown).

Failed settlement:

• Transaction: claim interview, credits.consumed: 0, retentionExpiresAt from failedAt.
• Release full reservation via settleCreditsInSession(..., consumed: 0).

Note: failureReason is in the message schema but not read by settleFailedInterview.

Database changes

Interview document (InterviewModel):

• settledAt, credits.consumed, retentionExpiresAt on successful claim.

Credits (via settleCreditsInSession in apps/api/src/modules/credits/credit.service.ts):

• Organization credit balances (reserved, available).
• Credit lots (remaining, reserved, status finalization).
• Ledger entries (SETTLE, release/forfeit entries as applicable) with reference.kind = INTERVIEW.

Synchronous interview status updates (COMPLETED / FAILED, timing, etc.) happen in the API before enqueue (bot-interview.service.ts), not in this consumer.

External service calls

None in the settlement path. Lambda is linked to resumeBucket for IAM read/write (infra/queues.ts); current settlement code does not call S3.

DLQ configuration

Setting	Value	Source
DLQ	interviewCompletedDlq.arn	infra/queues.ts
retry (max receives before DLQ)	3	infra/queues.ts dlq.retry
DLQ message retention	14 days	messageRetentionSeconds: 60 * 60 * 24 * 14

Retry behaviour

• Handler throw in handle: message id in batchItemFailures → SQS retries that message only (requires partialResponses: true).
• Invalid schema: no batchItemFailures → message deleted after error log.
• connectMongo failure: Lambda throws → entire batch retried.
• After 3 receives on primary queue: message moves to InterviewCompletedDlq (SST retry: 3).

Idempotent redelivery: duplicate settlement no-ops on settledAt claim; no second report.generate publish.

Visibility timeout

180 seconds (visibilityTimeoutSeconds: 180). Lambda timeout 30 seconds; comment requires ≥6× function timeout (infra/queues.ts).

Batch size

10 (batch.size: 10).

partialResponses

true — enables ReportBatchItemFailures on the event source mapping (infra/queues.ts).

Failure behaviour (summary)

Failure type	Outcome
Transient error in settlement	Per-message retry → DLQ after max receives
Malformed JSON / schema	Dropped (success from SQS’s view)
Mongo down at batch start	Whole batch retried
Report publish after settle fails	Settlement committed; error logged only

---

2. Report generate

Queue name

• SST component: ReportGenerate → reportGenerateQueue
• DLQ component: ReportGenerateDlq → reportGenerateDlq
• Deploy output: reportGenerateQueue, reportGenerateDlq (sst.config.ts)

Producer

Trigger	Service	Function
First successful completion settlement	Same API package (called from Lambda settlement path)	publishReportGenerate

Sources:

• apps/api/src/modules/credits/settlement.service.ts (after settleCompletedInterview claim succeeds)
• apps/api/src/modules/interviews/sqs.service.ts (publishReportGenerate)

Config: REPORT_GENERATE_QUEUE_URL. If missing, publish skipped with warning.

Message attributes:

Attribute	Type	Value
requestId	String	settlement-${interviewId} for settlement-triggered publishes
eventType	String	report.generate

Message schema / format

{
  "interviewId": "<ObjectId string>",
  "organizationId": "<ObjectId string>"
}

Validated by reportGenerateSchema in apps/functions/src/handlers/report-generate.ts.

Consumer (Lambda)

Property	Value
Function name	`${$app.name}-${$app.stage}-report-generate`
Handler	apps/functions/src/handlers/report-generate.handler
Business logic	generateReport in apps/api/src/modules/reports/report-generation.service.ts (re-export apps/functions/src/handlers/report-generation.ts)

Processing logic

1. Load interview (org-scoped, not soft-deleted).
2. If report already COMPLETED → return.
3. If attemptCount >= MAX_ATTEMPTS (3) → set status FAILED, return (no throw).
4. Atomic claim: status NOT_STARTED or FAILED → PROCESSING, increment attemptCount (upsert on interviewId).
5. If another worker holds PROCESSING → skip.
6. Short-circuit without LLM if duration < 60s or < 2 substantive answers → insufficient_evidence report.
7. Else OpenAI chat completion (gpt-4.1-mini, structured JSON schema) → persist full report, status COMPLETED.

Throws from OpenAI/parsing propagate to createSqsConsumer → SQS retry.

Admin retry API can call generateReport synchronously (retryReportGeneration); comment notes production retries should re-publish to SQS.

Database changes

InterviewReportModel (primary):

• Status lifecycle: NOT_STARTED / FAILED → PROCESSING → COMPLETED or FAILED
• Scores, recommendation, skill assessments, question analysis, generation metadata (model, tokens, promptVersion), retentionExpiresAt, etc.

Read-only:

• InterviewModel for snapshot, proctoring transcript, timing, credibility.

No interview or credit ledger updates in this consumer.

External service calls

• OpenAI — client.chat.completions.create with response_format: json_schema (report-generation.service.ts). Requires OPENAI_API_KEY on the Lambda (infra/queues.ts).

Lambda links resumeBucket; report generation reads interview.snapshot.resumeText and proctoring transcript from MongoDB, not S3 in the current service code.

DLQ configuration

Setting	Value	Source
DLQ	reportGenerateDlq.arn	infra/queues.ts
retry	3	infra/queues.ts
DLQ retention	14 days	same pattern as interview DLQ

Retry behaviour

• Same createSqsConsumer contract as interview queue.
• Document-level attemptCount and MAX_ATTEMPTS = 3 can mark report FAILED inside the handler without throwing (message acknowledged).
• SQS-level retries still apply when generateReport throws before that cap logic applies or on transient errors during LLM persistence.

Visibility timeout

1800 seconds (30 minutes). Lambda timeout 5 minutes; comment documents 6× rule (infra/queues.ts).

Batch size

1 (batch.size: 1).

partialResponses

true.

Failure behaviour (summary)

Failure type	Outcome
LLM / DB error during generation	Throw → per-message SQS retry → DLQ
Max document attempts reached	Report marked FAILED, handler returns success
Invalid message body	Dropped
Duplicate SQS delivery	Skip if COMPLETED; claim logic limits concurrent LLM work

---

Infrastructure cross-reference

Concern	File
Queue + subscriber definitions	infra/queues.ts
API env queue URLs	infra/api.ts
Deploy outputs	sst.config.ts
Producer SDK	apps/api/src/modules/interviews/sqs.service.ts
Consumer wrapper	apps/functions/src/platform/sqs-consumer.ts
Wrapper tests	apps/functions/src/platform/sqs-consumer.test.ts
Team rules	.cursor/rules/async-jobs.mdc, .cursor/rules/infra.mdc