Function createItemExtractor

services/session-ingest/src/queue-consumer.ts:43–161

(r2Key: string)

  env: Env,
  msg: IngestQueueMessage,
  ctx: ExecutionContext
): Promise<void> {
  const sessionRow = await loadSessionOrCleanupStaging(env, msg);
  if (!sessionRow) return;

  const body = await getStagingObjectBody(env, msg.r2Key);
  const mergedChanges = new Map<string, string | null>();
  const mergedAttentionSignals: AttentionSignal[] = [];

  try {
    const accepted = await ingestStagedSessionItems(
      env,
      msg,
      body,
      mergedChanges,
      mergedAttentionSignals
    );
    if (accepted) {
      await applyMetadataChanges(env, msg.kiloUserId, msg.sessionId, mergedChanges, ctx);
    }
  } catch (err) {
    // An earlier chunk may have committed to the DO before a later chunk (or the
    // JSON parse) failed. The DO reports a metadata change only when its stored
    // value differs, so on retry those already-persisted values won't be
    // re-emitted — Postgres would never catch up. Flush what we have now so the
    // two stores stay in sync. Best-effort: never mask the original error.
    await flushPartialMetadataChanges(env, msg, mergedChanges, ctx);
    // Same reasoning for attention signals: a committed status transition won't
    // re-emit on retry, so dispatch what was collected before the failure.
    scheduleAttentionSignalDispatch(
      env,
      msg,
      sessionRow,
      mergedChanges,
      mergedAttentionSignals,
      ctx
    );
    throw err;
  }

  scheduleAttentionSignalDispatch(env, msg, sessionRow, mergedChanges, mergedAttentionSignals, ctx);
  await env.SESSION_INGEST_R2.delete(msg.r2Key);
}

type IngestSessionRow = {
  session_id: string;
  parent_session_id: string | null;
  created_on_platform: string | null;
};

/**
 * Loads the session row for the queued message, or cleans up the staging object and returns
 * null if the session has been deleted since the message was queued. The parent column feeds
 * the attention-push eligibility check later in processing.
 */
async function loadSessionOrCleanupStaging(
  env: Env,
  msg: IngestQueueMessage