1import * as Effect from "effect/Effect";
2import * as Context from "effect/Context";
3import * as Layer from "effect/Layer";
4import * as Schema from "effect/Schema";
5import * as Clock from "effect/Clock";
6import type * as Cloudflare from "@cloudflare/workers-types";
7import type { Summary } from "./Summary.ts";
8
9interface Totals {
10 developer: string;
11 host: string;
12 platform: string;
13 arch: string;
14 ci: number;
15 local_ms: number;
16 unspecified_ms: number;
17 unspecified_hits: number;
18 remote_ms: number;
19 local_hits: number;
20 remote_hits: number;
21 executed: number;
22 restore_failures: number;
23}
24
25export class SavingsError extends Schema.TaggedError<SavingsError>()("SavingsError", {
26 operation: Schema.Literals(["record", "read", "cleanup"]),
27 cause: Schema.Defect(),
28}) {
29 override get message(): string {
30 return `Failed to ${this.operation} task-cache savings.`;
31 }
32}
33
34export class Database extends Context.Service<Database, Cloudflare.D1Database>()(
35 "@t3tools/task-cache/Savings/Database",
36) {}
37
38export class Savings extends Context.Service<
39 Savings,
40 {
41 readonly record: (
42 summary: typeof Summary.Type,
43 developer: string,
44 ) => Effect.Effect<void, SavingsError>;
45 readonly emitTotals: (developer?: string, host?: string) => Effect.Effect<void, SavingsError>;
46 readonly cleanup: Effect.Effect<void, SavingsError>;
47 }
48>()("@t3tools/task-cache/Savings") {}
49
50const make = Effect.gen(function* () {
51 const database = yield* Database;
52 const record = Effect.fn("TaskCache.persistSavings")(function* (
53 summary: typeof Summary.Type,
54 developer: string,
55 ) {
56 const local = summary.tasks.filter((task) => task.source === "local");
57 const unspecified = summary.tasks.filter((task) => task.source === "unspecified");
58 const remote = summary.tasks.filter((task) => task.source === "remote");
59 // D1 batch is transactional. changes() gates the increment on the preceding receipt insertion.
60 yield* Effect.tryPromise({
61 try: () =>
62 database.batch([
63 database
64 .prepare("INSERT INTO receipts(id, recorded_at) VALUES (?, ?) ON CONFLICT DO NOTHING")
65 .bind(summary.id, summary.recordedAt),
66 database
67 .prepare(`INSERT INTO savings(developer, host, platform, arch, ci, local_ms, remote_ms, local_hits, remote_hits, executed, restore_failures, unspecified_ms, unspecified_hits) SELECT ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? WHERE changes() = 1
68 ON CONFLICT DO UPDATE SET
69 local_ms = local_ms + excluded.local_ms, remote_ms = remote_ms + excluded.remote_ms,
70 local_hits = local_hits + excluded.local_hits, remote_hits = remote_hits + excluded.remote_hits,
71 executed = executed + excluded.executed, restore_failures = restore_failures + excluded.restore_failures,
72 unspecified_ms = unspecified_ms + excluded.unspecified_ms, unspecified_hits = unspecified_hits + excluded.unspecified_hits`)
73 .bind(
74 developer,
75 summary.host,
76 summary.platform,
77 summary.arch,
78 Number(summary.ci),
79 local.reduce((sum, task) => sum + task.savedMs, 0),
80 remote.reduce((sum, task) => sum + task.savedMs, 0),
81 local.length,
82 remote.length,
83 summary.tasks.filter((task) => task.source === "executed").length,
84 summary.tasks.filter((task) => task.source === "failed").length,
85 unspecified.reduce((sum, task) => sum + task.savedMs, 0),
86 unspecified.length,
87 ),
88 ]),
89 catch: (cause) => new SavingsError({ operation: "record", cause }),
90 });
91 });
92
93 const emitTotals = Effect.fn("TaskCache.emitTotals")(function* (
94 developer?: string,
95 host?: string,
96 ) {
97 const rows = yield* Effect.tryPromise({
98 try: () =>
99 (developer === undefined
100 ? database.prepare("SELECT * FROM savings")
101 : database
102 .prepare("SELECT * FROM savings WHERE developer = ? AND host = ?")
103 .bind(developer, host)
104 ).all<Totals>(),
105 catch: (cause) => new SavingsError({ operation: "read", cause }),
106 });
107 yield* Effect.forEach(rows.results, (row) =>
108 Effect.annotateCurrentSpan({
109 "developer.id": row.developer,
110 "host.name": row.host,
111 "host.os": row.platform,
112 "host.arch": row.arch,
113 ci: Boolean(row.ci),
114 "cache.local_saved_ms": row.local_ms,
115 "cache.unspecified_saved_ms": row.unspecified_ms,
116 "cache.unspecified_hits": row.unspecified_hits,
117 "cache.remote_saved_ms": row.remote_ms,
118 "cache.local_hits": row.local_hits,
119 "cache.remote_hits": row.remote_hits,
120 "cache.executed": row.executed,
121 "cache.restore_failures": row.restore_failures,
122 }).pipe(Effect.withSpan("TaskCache.lifetime")),
123 );
124 });
125
126 const cleanup = Effect.gen(function* () {
127 const now = yield* Clock.currentTimeMillis;
128 yield* Effect.tryPromise({
129 try: () =>
130 database
131 .prepare("DELETE FROM receipts WHERE recorded_at < ?")
132 .bind(now - 31 * 86400_000)
133 .run(),
134 catch: (cause) => new SavingsError({ operation: "cleanup", cause }),
135 });
136 });
137 return Savings.of({ record, emitTotals, cleanup });
138});
139
140export const layer = Layer.effect(Savings, make);