Submatter/apps/web/server/lib/convoy/convoy-service.ts

import { and, desc, eq, sql } from "drizzle-orm";

import type { db as dbInstance } from "@tasks/database";
import { convoyEvents, objects, workerCheckpoints } from "@tasks/database/schema";
import {
  convoyObjectContentSchema,
  convoyStatusSchema,
  validateConvoyTransition,
  type ConvoyObjectContent,
  type ConvoyStatus,
} from "@tasks/shared/convoy-lifecycle";

type Db = typeof dbInstance;
type ObjectRow = typeof objects.$inferSelect;

export type ConvoySummary = {
  id: string;
  title: string;
  status: ConvoyStatus;
  description: string | null;
  content: ConvoyObjectContent | null;
  updatedAt: Date;
};

function parseStatus(raw: string | null): ConvoyStatus {
  const parsed = convoyStatusSchema.safeParse(raw ?? "draft");
  return parsed.success ? parsed.data : "draft";
}

function toConvoySummary(row: ObjectRow): ConvoySummary {
  const contentParsed = convoyObjectContentSchema.safeParse(row.content);
  return {
    id: row.id,
    title: row.title,
    status: parseStatus(row.status),
    description: row.description,
    content: contentParsed.success ? contentParsed.data : null,
    updatedAt: row.updatedAt,
  };
}

export function convoyProjectsFilter(workspaceId: string) {
  return and(
    eq(objects.workspaceId, workspaceId),
    eq(objects.type, "project"),
    sql`${objects.content} ->> 'convoySlug' IS NOT NULL`,
  );
}

export async function listConvoyProjects(db: Db, workspaceId: string): Promise<ConvoySummary[]> {
  const rows = await db
    .select()
    .from(objects)
    .where(convoyProjectsFilter(workspaceId))
    .orderBy(desc(objects.updatedAt));

  return rows
    .map(toConvoySummary)
    .filter((row) => row.content !== null && row.status !== "cancelled");
}

export async function findConvoyBySlug(db: Db, workspaceId: string, slug: string) {
  const [row] = await db
    .select()
    .from(objects)
    .where(
      and(
        convoyProjectsFilter(workspaceId),
        sql`${objects.content} ->> 'convoySlug' = ${slug}`,
      ),
    )
    .limit(1);
  return row ?? null;
}

export async function loadConvoyProject(db: Db, workspaceId: string, convoyId: string) {
  const [row] = await db
    .select()
    .from(objects)
    .where(
      and(
        eq(objects.id, convoyId),
        convoyProjectsFilter(workspaceId),
      ),
    )
    .limit(1);
  return row ?? null;
}

export async function insertConvoyEvent(
  db: Db,
  input: {
    workspaceId: string;
    convoyId: string;
    briefId?: string | null;
    role: string;
    actor?: string | null;
    eventType: string;
    metadata?: Record<string, unknown> | null;
  },
) {
  const [row] = await db
    .insert(convoyEvents)
    .values({
      workspaceId: input.workspaceId,
      convoyId: input.convoyId,
      briefId: input.briefId ?? null,
      role: input.role,
      actor: input.actor ?? null,
      eventType: input.eventType,
      metadata: input.metadata ?? null,
    })
    .returning();
  return row;
}

export async function transitionConvoy(
  db: Db,
  input: {
    workspaceId: string;
    convoyId: string;
    toStatus: ConvoyStatus;
    actor: string;
    reason?: string;
  },
): Promise<{ convoy: ConvoySummary; from: ConvoyStatus; to: ConvoyStatus }> {
  const convoy = await loadConvoyProject(db, input.workspaceId, input.convoyId);
  if (!convoy) {
    throw new Error("Convoy not found");
  }

  const fromStatus = parseStatus(convoy.status);
  validateConvoyTransition({
    from: fromStatus,
    to: input.toStatus,
    actor: input.actor,
    reason: input.reason,
  });

  const [updated] = await db
    .update(objects)
    .set({
      status: input.toStatus,
      updatedAt: new Date(),
    })
    .where(eq(objects.id, convoy.id))
    .returning();

  if (!updated) {
    throw new Error("Failed to update convoy status");
  }

  await insertConvoyEvent(db, {
    workspaceId: input.workspaceId,
    convoyId: convoy.id,
    role: input.actor,
    actor: input.actor,
    eventType: "status_change",
    metadata: {
      from: fromStatus,
      to: input.toStatus,
      reason: input.reason ?? null,
    },
  });

  return {
    convoy: toConvoySummary(updated),
    from: fromStatus,
    to: input.toStatus,
  };
}

export async function getConvoyBriefs(db: Db, workspaceId: string, convoyId: string) {
  return db
    .select({
      id: objects.id,
      title: objects.title,
      status: objects.status,
      sortOrder: objects.sortOrder,
    })
    .from(objects)
    .where(and(eq(objects.parentId, convoyId), eq(objects.workspaceId, workspaceId)))
    .orderBy(objects.sortOrder);
}

export async function getRecentConvoyEvents(
  db: Db,
  workspaceId: string,
  convoyId: string,
  limit: number,
) {
  return db
    .select()
    .from(convoyEvents)
    .where(and(eq(convoyEvents.convoyId, convoyId), eq(convoyEvents.workspaceId, workspaceId)))
    .orderBy(desc(convoyEvents.occurredAt))
    .limit(limit);
}

export async function getConvoyPrUrl(
  db: Db,
  workspaceId: string,
  convoyId: string,
): Promise<{ prUrl: string | null; prNumber: number | null }> {
  const events = await db
    .select({ metadata: convoyEvents.metadata })
    .from(convoyEvents)
    .where(and(eq(convoyEvents.convoyId, convoyId), eq(convoyEvents.workspaceId, workspaceId)))
    .orderBy(desc(convoyEvents.occurredAt))
    .limit(30);

  for (const row of events) {
    const meta = row.metadata as Record<string, unknown> | null;
    if (meta && typeof meta.prUrl === "string" && meta.prUrl) {
      const prNumber = typeof meta.prNumber === "number" ? meta.prNumber : null;
      return { prUrl: meta.prUrl, prNumber };
    }
  }

  const convoy = await loadConvoyProject(db, workspaceId, convoyId);
  if (convoy?.content && typeof convoy.content === "object" && convoy.content !== null) {
    const content = convoy.content as Record<string, unknown>;
    if (typeof content.prUrl === "string" && content.prUrl) {
      const prNumber = typeof content.prNumber === "number" ? content.prNumber : null;
      return { prUrl: content.prUrl, prNumber };
    }
    const parsed = convoyObjectContentSchema.safeParse(convoy.content);
    if (parsed.success && parsed.data.prUrl) {
      return { prUrl: parsed.data.prUrl, prNumber: parsed.data.prNumber ?? null };
    }
  }

  return { prUrl: null, prNumber: null };
}

/** Checkpoint count + last time for mission-control detail (workspace-scoped). */
export async function getConvoyCheckpointSummary(
  db: Db,
  workspaceId: string,
  convoyId: string,
): Promise<{ count: number; lastOccurredAt: Date | null }> {
  const [row] = await db
    .select({
      count: sql<number>`count(*)::int`,
      lastOccurredAt: sql<Date | null>`max(${workerCheckpoints.occurredAt})`,
    })
    .from(workerCheckpoints)
    .where(
      and(
        eq(workerCheckpoints.workspaceId, workspaceId),
        eq(workerCheckpoints.convoyId, convoyId),
      ),
    );

  return {
    count: row?.count ?? 0,
    lastOccurredAt: row?.lastOccurredAt ?? null,
  };
}