Getting journal through SQL

getJournal.ts

// ---------------------------------------------------------------------------
// Journal data model
// ---------------------------------------------------------------------------

type Bytes = number[];
type CompletionId = number;
type MillisSinceEpoch = number;

interface Header {
  name: string;
  value: string;
}

interface Failure {
  code: number;
  message: string;
  metadata?: { key: string; value: string }[];
}

type SignalResult = "Void" | { Success: Bytes } | { Failure: Failure };

type InvocationTarget =
  | { Service: { name: string; handler: string; scope: string | null } }
  | {
      VirtualObject: {
        name: string;
        key: string;
        handler: string;
        scope: string | null;
      };
    }
  | {
      Workflow: {
        name: string;
        key: string;
        handler: string;
        scope: string | null;
      };
    };

interface CallRequest {
  invocation_id: string;
  invocation_target: InvocationTarget;
  parameter: Bytes;
  headers: Header[];
  idempotency_key?: string;
  limit_key?: string;
}

type AttachInvocationTarget =
  | { InvocationId: string }
  | { IdempotentRequest: {
    service_name: string;
    service_key: string | null;
    service_handler: string;
    idempotency_key: string;
    scope: string | null;
  } }
  | { Workflow: {
    service_name: string;
    key: string;
    scope: string | null;
  } };

type SignalId = { Index: number } | { Name: string };

// Commands - crates/types/src/journal_v2/command.rs
type Command =
  | { Input: { headers: Header[]; payload: Bytes; name: string } }
  | {
      Output: { result: { Success: Bytes } | { Failure: Failure }; name: string };
    }
  | { GetLazyState: { key: string; completion_id: CompletionId; name: string } }
  | { SetState: { key: string; value: Bytes; name: string } }
  | { ClearState: { key: string; name: string } }
  | { ClearAllState: { name: string } }
  | { GetLazyStateKeys: { completion_id: CompletionId; name: string } }
  | { GetEagerState: { key: string; result: "Void" | { Success: Bytes }; name: string } }
  | { GetEagerStateKeys: { state_keys: string[]; name: string } }
  | { GetPromise: { key: string; completion_id: CompletionId; name: string } }
  | { PeekPromise: { key: string; completion_id: CompletionId; name: string } }
  | {
      CompletePromise: {
        key: string;
        value: { Success: Bytes } | { Failure: Failure };
        completion_id: CompletionId;
        name: string;
      };
    }
  | {
      Sleep: {
        wake_up_time: MillisSinceEpoch;
        completion_id: CompletionId;
        name: string;
      };
    }
  | {
      Call: {
        request: CallRequest;
        invocation_id_completion_id: CompletionId;
        result_completion_id: CompletionId;
        name: string;
      };
    }
  | {
      OneWayCall: {
        request: CallRequest;
        invoke_time: MillisSinceEpoch;
        invocation_id_completion_id: CompletionId;
        name: string;
      };
    }
  | {
      SendSignal: {
        target_invocation_id: string;
        signal_id: SignalId;
        result: SignalResult;
        name: string;
      };
    }
  | { Run: { completion_id: CompletionId; name: string } }
  | {
      AttachInvocation: {
        target: AttachInvocationTarget;
        completion_id: CompletionId;
        name: string;
      };
    }
  | {
      GetInvocationOutput: {
        target: AttachInvocationTarget;
        completion_id: CompletionId;
        name: string;
      };
    }
  | {
      CompleteAwakeable: {
        id: string;
        result: { Success: Bytes } | { Failure: Failure };
        name: string;
      };
    };

// Notifications - crates/types/src/journal_v2/notification.rs
type Completion =
  | { GetLazyState: { completion_id: CompletionId; result: "Void" | { Success: Bytes } } }
  | {
      GetLazyStateKeys: { completion_id: CompletionId; state_keys: string[] };
    }
  | {
      GetPromise: {
        completion_id: CompletionId;
        result: { Success: Bytes } | { Failure: Failure };
      };
    }
  | {
      PeekPromise: {
        completion_id: CompletionId;
        result: "Void" | { Success: Bytes } | { Failure: Failure };
      };
    }
  | {
      CompletePromise: {
        completion_id: CompletionId;
        result: "Void" | { Failure: Failure };
      };
    }
  | { Sleep: { completion_id: CompletionId } }
  | {
      CallInvocationId: {
        completion_id: CompletionId;
        invocation_id: string;
      };
    }
  | {
      Call: {
        completion_id: CompletionId;
        result: { Success: Bytes } | { Failure: Failure };
      };
    }
  | {
      Run: {
        completion_id: CompletionId;
        result: { Success: Bytes } | { Failure: Failure };
      };
    }
  | {
      AttachInvocation: {
        completion_id: CompletionId;
        result: { Success: Bytes } | { Failure: Failure };
      };
    }
  | {
      GetInvocationOutput: {
        completion_id: CompletionId;
        result: "Void" | { Success: Bytes } | { Failure: Failure };
      };
    };

type Notification =
  | { Completion: Completion }
  | { Signal: { id: SignalId; result: SignalResult } };

type JournalEntry = { Command: Command } | { Notification: Notification };

function isCommand<K extends string>(
  entry: JournalEntry,
  variant: K
): entry is { Command: Extract<Command, Record<K, unknown>> } {
  return "Command" in entry && variant in entry.Command;
}

// ---------------------------------------------------------------------------
// getJournal function
// ---------------------------------------------------------------------------

// Fetch the journal of an invocation via the SQL introspection API.
//
// Returns the ordered parsed journal entries.
async function getJournal(
  adminUrl: string,
  invocationId: string
): Promise<JournalEntry[]> {
  const res = await fetch(`${adminUrl}/query`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Accept: "application/json",
    },
    body: JSON.stringify({
      query: `SELECT entry_json FROM sys_journal WHERE id = '${invocationId}' ORDER BY index`,
    }),
  });
  if (!res.ok) {
    throw new Error(`Journal query failed: ${res.status} ${await res.text()}`);
  }
  const { rows } = (await res.json()) as { rows: { entry_json: string }[] };
  return rows.map((row) => JSON.parse(row.entry_json) as JournalEntry);
}