acp

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122// ACP registry lifecycle: register cached, known and custom agents after
// loading their marks, and refresh the daily registry on an hourly timer.
import * as Effect from "effect/Effect";
import * as Fiber from "effect/Fiber";
import * as Queue from "effect/Queue";
import * as Scope from "effect/Scope";
import * as Exit from "effect/Exit";
import { Agents } from "convergence/effect";
import type { PluginServices } from "convergence/effect";
import type { RpcTransport } from "../sdk/effect.ts";
import { AcpAgent } from "./agent.ts";
import {
  MAX_CACHE_AGE,
  adoptLegacy,
  assemble,
  fetchIcons,
  loadCustom,
  loadIcons,
  loadRegistry,
  refreshRegistry,
} from "./registry.ts";
import type { Entry, Custom, Icons } from "./registry.ts";
import type { AcpContext } from "./types.ts";
const REFRESH_CHECK = 60 * 60 * 1000;
export class AcpPlugin {
  agents = new Map<string, AcpAgent>();
  registrations = new Map<string, Scope.Closeable>();
  icons: Icons = {};
  custom: Custom[] = [];
  fetchedAt = 0;
  refreshing: Fiber.Fiber<void, unknown> | null = null;
  timer: Fiber.Fiber<never, unknown> | null = null;
  scope: Scope.Scope;
  transport: RpcTransport["Service"];
  context: AcpContext;
  constructor(scope: Scope.Scope, transport: RpcTransport["Service"], context: AcpContext) {
    this.scope = scope;
    this.transport = transport;
    this.context = context;
    this.context.icons = this.icons;
  }
  start = Effect.fn("Acp.start")(function* (this: AcpPlugin, now = Date.now()) {
    const [loaded, icons, custom] = yield* Effect.all(
      [loadRegistry(now), loadIcons(), adoptLegacy().pipe(Effect.andThen(loadCustom()))],
      { concurrency: "unbounded" },
    );
    Object.assign(this.icons, icons);
    this.custom = custom;
    this.fetchedAt = loaded.stale ? 0 : now;
    const entries = assemble(loaded.registry, custom);
    yield* fetchIcons(entries, this.icons);
    yield* this.register(entries);
    if (loaded.stale) yield* this.beginRefresh();
    this.timer = yield* Effect.forever(
      Effect.sleep(REFRESH_CHECK).pipe(
        Effect.andThen(
          Effect.suspend(() => (Date.now() - this.fetchedAt >= MAX_CACHE_AGE ? this.beginRefresh() : Effect.void)),
        ),
      ),
    ).pipe(Effect.forkIn(this.scope));
  });
  register = Effect.fn("Acp.register")(function* (this: AcpPlugin, entries: Entry[]) {
    const listed = new Set(entries.map((entry) => entry.id));
    for (const [id, agent] of [...this.agents]) {
      if (listed.has(id) || agent.entry.source.kind !== "registry") continue;
      const registration = this.registrations.get(id);
      if (registration) yield* Scope.close(registration, Exit.void);
      this.registrations.delete(id);
      this.agents.delete(id);
    }
    const agents = yield* Agents;
    for (const entry of entries) {
      const known = this.agents.get(entry.id);
      if (known) {
        known.setEntry(entry);
        continue;
      }
      const scope = yield* Scope.fork(this.scope, "sequential");
      yield* Effect.gen({ self: this }, function* () {
        const jobs = yield* Effect.acquireRelease(
          Queue.make<Effect.Effect<unknown, unknown, PluginServices>>(),
          Queue.shutdown,
        );
        const agent = new AcpAgent({ entry, context: this.context, scope, transport: this.transport, jobs });
        const handle = yield* agents.register(agent.definition());
        agent.emit = handle.emit;
        this.agents.set(entry.id, agent);
        this.registrations.set(entry.id, scope);
        yield* agent.background();
      }).pipe(Scope.provide(scope));
    }
  });
  beginRefresh = Effect.fn("Acp.beginRefresh")(function* (this: AcpPlugin, now = Date.now()) {
    if (this.refreshing) return;
    this.refreshing = yield* this.refreshEntries(now).pipe(
      Effect.ensuring(
        Effect.sync(() => {
          this.refreshing = null;
        }),
      ),
      Effect.forkIn(this.scope),
    );
  });
  refresh = Effect.fn("Acp.refresh")(function* (this: AcpPlugin, now = Date.now()) {
    yield* this.beginRefresh(now);
    const refreshing = this.refreshing;
    if (refreshing) yield* Fiber.join(refreshing);
  });
  refreshEntries = Effect.fn("Acp.refreshEntries")(function* (this: AcpPlugin, now = Date.now()) {
    const registry = yield* refreshRegistry(now);
    if (!registry) return;
    this.fetchedAt = now;
    this.custom = yield* loadCustom();
    const entries = assemble(registry, this.custom);
    yield* fetchIcons(entries, this.icons, { force: true });
    yield* this.register(entries);
  });
  stop = Effect.fn("Acp.stop")(function* (this: AcpPlugin) {
    if (this.timer) yield* Fiber.interrupt(this.timer);
    this.timer = null;
  });
}