WRNexusJS/packages/queue/src/subject.ts

import type { Context } from "@wrnexus/core";
import { exportSubjectContext, importSubjectContext, type SubjectContext } from "@wrnexus/rpc";
import type { AddOptions, Job, Queue } from "./index.ts";
import type { DurableQueue } from "./durable.ts";

const QUEUE_AUDIENCE = "wrnexus-queue";
interface SubjectEnvelope<T> {
  payload: T;
  identity?: string;
}

export interface SubjectJob<T> extends Omit<Job<SubjectEnvelope<T>>, "data"> {
  data: T;
  subject?: SubjectContext;
}

export interface SubjectQueue {
  add<T>(
    ctx: Context,
    name: string,
    data: T,
    options?: AddOptions,
  ): Promise<Job<SubjectEnvelope<T>>>;
  process<T>(
    name: string,
    handler: (job: SubjectJob<T>, context: { signal: AbortSignal }) => void | Promise<void>,
  ): void;
}

/** Queue adapter that persists a signed end-user context alongside job data. */
/** Works with both the in-memory Queue and createDurableQueue(). */
export function subjectQueue(queue: Queue | DurableQueue): SubjectQueue {
  return {
    async add(ctx, name, data, options) {
      const identity = await exportSubjectContext(ctx, QUEUE_AUDIENCE);
      return queue.add(name, { payload: data, ...(identity ? { identity } : {}) }, options);
    },
    process(name, handler) {
      queue.process<SubjectEnvelope<unknown>>(name, async (job, context) => {
        const envelope = job.data;
        if (!envelope || typeof envelope !== "object" || !("payload" in envelope)) return;
        let subject: SubjectContext | undefined;
        if (envelope.identity !== undefined) {
          if (typeof envelope.identity !== "string") return;
          try {
            subject = await importSubjectContext(envelope.identity, QUEUE_AUDIENCE);
          } catch {
            return;
          }
        }
        await handler(
          { ...job, data: envelope.payload as never, ...(subject ? { subject } : {}) },
          context,
        );
      });
    },
  };
}