zopu-code/repos/daemon/src/convex.ts

import { api } from "@code/backend/convex/_generated/api";
import type { Id } from "@code/backend/convex/_generated/dataModel";
import { env } from "@code/env/server";
import { ConvexClient } from "convex/browser";
import { Cause, Context, Effect, Layer, Queue, Schema, Stream } from "effect";

export class ConvexControlPlaneError extends Schema.TaggedErrorClass<ConvexControlPlaneError>()(
  "ConvexControlPlaneError",
  {
    operation: Schema.String,
    cause: Schema.Defect(),
  }
) {}

export interface DaemonCommand {
  readonly _id: Id<"daemonCommands">;
  readonly daemonId: string;
  readonly actorKey: readonly string[];
  readonly method: string;
  readonly args: unknown;
}

const tryPromise = <A>(operation: string, evaluate: () => Promise<A>) =>
  Effect.tryPromise({
    try: evaluate,
    catch: (cause) => new ConvexControlPlaneError({ operation, cause }),
  });

export class ConvexControlPlane extends Context.Service<
  ConvexControlPlane,
  {
    readonly commands: Stream.Stream<
      readonly DaemonCommand[],
      ConvexControlPlaneError
    >;
    readonly connect: (input: {
      readonly sessionId: string;
      readonly hostname: string;
      readonly platform: string;
      readonly architecture: string;
    }) => Effect.Effect<void, ConvexControlPlaneError>;
    readonly heartbeat: (
      sessionId: string
    ) => Effect.Effect<void, ConvexControlPlaneError>;
    readonly disconnect: (
      sessionId: string
    ) => Effect.Effect<void, ConvexControlPlaneError>;
    readonly claim: (
      commandId: Id<"daemonCommands">,
      sessionId: string
    ) => Effect.Effect<DaemonCommand | null, ConvexControlPlaneError>;
    readonly start: (
      commandId: Id<"daemonCommands">,
      sessionId: string
    ) => Effect.Effect<boolean, ConvexControlPlaneError>;
    readonly succeed: (
      commandId: Id<"daemonCommands">,
      sessionId: string,
      result: unknown
    ) => Effect.Effect<boolean, ConvexControlPlaneError>;
    readonly fail: (
      commandId: Id<"daemonCommands">,
      sessionId: string,
      error: string
    ) => Effect.Effect<boolean, ConvexControlPlaneError>;
    readonly recordEvent: (input: {
      readonly commandId?: Id<"daemonCommands">;
      readonly kind: string;
      readonly data: unknown;
    }) => Effect.Effect<void, ConvexControlPlaneError>;
    readonly close: Effect.Effect<void, ConvexControlPlaneError>;
  }
>()("@code/daemon/ConvexControlPlane") {
  static readonly layer = Layer.effect(
    ConvexControlPlane,
    Effect.gen(function* () {
      const client = yield* Effect.acquireRelease(
        Effect.sync(() => new ConvexClient(env.CONVEX_URL)),
        (client) => Effect.promise(() => client.close())
      );
      const commands = Stream.callback<
        readonly DaemonCommand[],
        ConvexControlPlaneError
      >((queue) =>
        Effect.acquireRelease(
          Effect.sync(() =>
            client.onUpdate(
              api.daemonCommands.available,
              { daemonId: env.DAEMON_ID },
              (items) => {
                Queue.offerUnsafe(queue, items);
              },
              (cause) => {
                Queue.failCauseUnsafe(
                  queue,
                  Cause.fail(
                    new ConvexControlPlaneError({
                      operation: "subscribe commands",
                      cause,
                    })
                  )
                );
              }
            )
          ),
          (unsubscribe) => Effect.sync(unsubscribe)
        )
      );

      return ConvexControlPlane.of({
        commands,
        connect: (input) =>
          tryPromise("connect daemon", async () => {
            await client.mutation(api.daemonRuntime.connect, {
              daemonId: env.DAEMON_ID,
              version: env.DAEMON_VERSION,
              ...input,
            });
          }),
        heartbeat: (sessionId) =>
          tryPromise("heartbeat daemon", async () => {
            await client.mutation(api.daemonRuntime.heartbeat, {
              daemonId: env.DAEMON_ID,
              sessionId,
            });
          }),
        disconnect: (sessionId) =>
          tryPromise("disconnect daemon", async () => {
            await client.mutation(api.daemonRuntime.disconnect, {
              daemonId: env.DAEMON_ID,
              sessionId,
            });
          }),
        claim: (commandId, sessionId) =>
          tryPromise("claim command", () =>
            client.mutation(api.daemonCommands.claim, {
              commandId,
              daemonId: env.DAEMON_ID,
              sessionId,
              leaseDurationMs: env.DAEMON_COMMAND_LEASE_MS,
            })
          ),
        start: (commandId, sessionId) =>
          tryPromise("start command", () =>
            client.mutation(api.daemonCommands.start, { commandId, sessionId })
          ),
        succeed: (commandId, sessionId, result) =>
          tryPromise("complete command", () =>
            client.mutation(api.daemonCommands.succeed, {
              commandId,
              sessionId,
              result,
            })
          ),
        fail: (commandId, sessionId, error) =>
          tryPromise("fail command", () =>
            client.mutation(api.daemonCommands.fail, {
              commandId,
              sessionId,
              error,
            })
          ),
        recordEvent: (input) =>
          tryPromise("record daemon event", async () => {
            await client.mutation(api.daemonRuntime.recordEvent, {
              daemonId: env.DAEMON_ID,
              ...input,
            });
          }),
        close: tryPromise("close Convex client", () => client.close()),
      });
    })
  );
}