Build a Claimable Community DM Queue That Cannot Silently Drop Requests

Member requests private help
        ↓
Request enters queue
        ↓
One coordinator claims it
        ↓
Worker attempts to open/send the first DM
        ↓
Conversation becomes active or enters recovery
        ↓
Coordinator resolves the request

mkdir community-dm-queue
cd community-dm-queue
npm init -y
npm install --save-dev typescript tsx @types/node
mkdir -p src test

{
  "type": "module",
  "scripts": {
    "test": "tsx --test test/*.test.ts"
  },
  "devDependencies": {
    "@types/node": "latest",
    "tsx": "latest",
    "typescript": "latest"
  }
}

export type RequestState =
  | { kind: "queued"; attempts: number }
  | {
      kind: "claimed";
      attempts: number;
      ownerId: string;
      leaseUntil: number;
    }
  | {
      kind: "opening";
      attempts: number;
      ownerId: string;
      deliveryKey: string;
      leaseUntil: number;
    }
  | {
      kind: "needs_reconcile";
      attempts: number;
      ownerId: string;
      deliveryKey: string;
      reason: string;
    }
  | {
      kind: "active";
      attempts: number;
      ownerId: string;
      conversationId: string;
    }
  | {
      kind: "resolved";
      ownerId: string;
      conversationId: string;
      resolution: string;
    };

export interface DmRequest {
  id: string;
  memberId: string;
  topic: string;
  sourceLocale?: string;
  revision: number;
  state: RequestState;
}

export type Event =
  | { type: "claim"; ownerId: string; leaseUntil: number }
  | { type: "start_open"; ownerId: string; deliveryKey: string; leaseUntil: number }
  | { type: "dm_opened"; deliveryKey: string; conversationId: string }
  | { type: "delivery_failed"; deliveryKey: string; definitive: boolean; reason: string }
  | { type: "lease_elapsed"; now: number }
  | { type: "reconciled_absent"; deliveryKey: string }
  | { type: "resolve"; ownerId: string; resolution: string };

export function transition(request: DmRequest, event: Event): DmRequest {
  const state = request.state;
  let next: RequestState;

  switch (event.type) {
    case "claim": {
      if (state.kind !== "queued") throw new Error("request_not_available");

      next = {
        kind: "claimed",
        attempts: state.attempts,
        ownerId: event.ownerId,
        leaseUntil: event.leaseUntil
      };
      break;
    }

    case "start_open": {
      if (state.kind !== "claimed" || state.ownerId !== event.ownerId) {
        throw new Error("claim_not_owned");
      }

      next = {
        kind: "opening",
        attempts: state.attempts + 1,
        ownerId: state.ownerId,
        deliveryKey: event.deliveryKey,
        leaseUntil: event.leaseUntil
      };
      break;
    }

    case "dm_opened": {
      if (
        (state.kind !== "opening" && state.kind !== "needs_reconcile") ||
        state.deliveryKey !== event.deliveryKey
      ) {
        throw new Error("stale_delivery_result");
      }

      next = {
        kind: "active",
        attempts: state.attempts,
        ownerId: state.ownerId,
        conversationId: event.conversationId
      };
      break;
    }

    case "delivery_failed": {
      if (state.kind !== "opening" || state.deliveryKey !== event.deliveryKey) {
        throw new Error("stale_delivery_failure");
      }

      next = event.definitive
        ? { kind: "queued", attempts: state.attempts }
        : {
            kind: "needs_reconcile",
            attempts: state.attempts,
            ownerId: state.ownerId,
            deliveryKey: state.deliveryKey,
            reason: event.reason
          };
      break;
    }

    case "lease_elapsed": {
      if (state.kind === "claimed" && event.now >= state.leaseUntil) {
        next = { kind: "queued", attempts: state.attempts };
        break;
      }

      if (state.kind === "opening" && event.now >= state.leaseUntil) {
        next = {
          kind: "needs_reconcile",
          attempts: state.attempts,
          ownerId: state.ownerId,
          deliveryKey: state.deliveryKey,
          reason: "delivery_outcome_unknown"
        };
        break;
      }

      throw new Error("lease_not_elapsed");
    }

    case "reconciled_absent": {
      if (
        state.kind !== "needs_reconcile" ||
        state.deliveryKey !== event.deliveryKey
      ) {
        throw new Error("reconciliation_mismatch");
      }

      next = { kind: "queued", attempts: state.attempts };
      break;
    }

    case "resolve": {
      if (state.kind !== "active" || state.ownerId !== event.ownerId) {
        throw new Error("active_request_not_owned");
      }

      next = {
        kind: "resolved",
        ownerId: state.ownerId,
        conversationId: state.conversationId,
        resolution: event.resolution
      };
      break;
    }
  }

  return {
    ...request,
    revision: request.revision + 1,
    state: next
  };
}

export interface RequestRepository {
  get(id: string): Promise<DmRequest | undefined>;

  saveIfRevisionMatches(
    request: DmRequest,
    expectedRevision: number
  ): Promise<boolean>;
}

export async function applyEvent(
  repository: RequestRepository,
  requestId: string,
  event: Event
): Promise<DmRequest> {
  const current = await repository.get(requestId);
  if (!current) throw new Error("request_not_found");

  const updated = transition(current, event);
  const saved = await repository.saveIfRevisionMatches(
    updated,
    current.revision
  );

  if (!saved) throw new Error("concurrent_update");
  return updated;
}

UPDATE dm_requests
SET state_json = ?, revision = revision + 1
WHERE id = ? AND revision = ?;

export interface MessagingPort {
  deliverFirstMessage(input: {
    deliveryKey: string;
    memberId: string;
    coordinatorId: string;
    text: string;
  }): Promise<{ conversationId: string }>;
}

CREATE TABLE delivery_ledger (
  delivery_key TEXT PRIMARY KEY,
  status TEXT NOT NULL,
  conversation_id TEXT,
  updated_at TEXT NOT NULL
);

interface StoredMessage {
  messageId: string;
  conversationId: string;
  authorId: string;
  originalText: string;
  originalLocale?: string;
}

interface TranslationView {
  messageId: string;
  targetLocale: string;
  status: "requested" | "ready" | "failed";
  translatedText?: string;
  failureCode?: string;
}

import assert from "node:assert/strict";
import test from "node:test";
import { DmRequest, transition } from "../src/domain.js";

function queued(): DmRequest {
  return {
    id: "req-1",
    memberId: "member-7",
    topic: "account-specific question",
    revision: 0,
    state: { kind: "queued", attempts: 0 }
  };
}

test("a second coordinator cannot claim an owned request", () => {
  const claimed = transition(queued(), {
    type: "claim",
    ownerId: "coordinator-a",
    leaseUntil: 1_000
  });

  assert.throws(
    () => transition(claimed, {
      type: "claim",
      ownerId: "coordinator-b",
      leaseUntil: 2_000
    }),
    /request_not_available/
  );
});

test("an opening timeout requires reconciliation, not blind retry", () => {
  const claimed = transition(queued(), {
    type: "claim",
    ownerId: "coordinator-a",
    leaseUntil: 1_000
  });

  const opening = transition(claimed, {
    type: "start_open",
    ownerId: "coordinator-a",
    deliveryKey: "delivery-123",
    leaseUntil: 2_000
  });

  const uncertain = transition(opening, {
    type: "lease_elapsed",
    now: 2_001
  });

  assert.equal(uncertain.state.kind, "needs_reconcile");
});

test("a matching late result can recover an uncertain delivery", () => {
  const request: DmRequest = {
    ...queued(),
    state: {
      kind: "needs_reconcile",
      attempts: 1,
      ownerId: "coordinator-a",
      deliveryKey: "delivery-123",
      reason: "delivery_outcome_unknown"
    }
  };

  const active = transition(request, {
    type: "dm_opened",
    deliveryKey: "delivery-123",
    conversationId: "conversation-9"
  });

  assert.equal(active.state.kind, "active");
});

test("a stale callback cannot activate the wrong request attempt", () => {
  const request: DmRequest = {
    ...queued(),
    state: {
      kind: "needs_reconcile",
      attempts: 2,
      ownerId: "coordinator-b",
      deliveryKey: "delivery-new",
      reason: "delivery_outcome_unknown"
    }
  };

  assert.throws(
    () => transition(request, {
      type: "dm_opened",
      deliveryKey: "delivery-old",
      conversationId: "conversation-old"
    }),
    /stale_delivery_result/
  );
});

npm test