clover's git

the-knife

1	import * as Fly from "@/Fly";
2
3	export const Cache = Fly.Redis("WorkerQueue", { eviction: false });
4	export const LedgerSite = Fly.App("LedgerSite");
5	export const WorkerSite = Fly.App("WorkerSite");
6
7	export const services: Fly.MachineService[] = [
8	{
9	protocol: "tcp",
10	internalPort: 3000,
11	autostop: "off",
12	ports: [{ port: 443, handlers: ["tls", "http"] }],
13	checks: [
14	{
15	type: "http",
16	port: 3000,
17	path: "/health",
18	interval: "2s",
19	timeout: "1s",
20	},
21	],
22	},
23	];
24
25	export const scripts = {
26	event: `
27	local t=redis.call('TIME')
28	local e=cjson.decode(ARGV[1]); e.at=tonumber(t[1])*1000+math.floor(tonumber(t[2])/1000)
29	redis.call('RPUSH','bg:events',cjson.encode(e)); return e.at`,
30	enqueue: `
31	if redis.call('SADD','bg:produced',ARGV[1])==0 then return 0 end
32	return redis.call('XADD','bg:jobs','*','job',ARGV[1],'kind',ARGV[2])`,
33	pause: `return redis.call('SET','bg:producer-paused','1')`,
34	tick: `
35	if redis.call('EXISTS','bg:producer-paused')==1 then return 0 end
36	local t=redis.call('TIME'); local slot='tick:'..t[1]; local fresh=redis.call('SADD','bg:produced',slot)
37	if fresh==1 then redis.call('XADD','bg:jobs','*','job',slot,'kind','quick') end
38	redis.call('RPUSH','bg:events',cjson.encode({event='producer',machine=ARGV[1],version=ARGV[2],job=slot,fresh=fresh,at=tonumber(t[1])*1000+math.floor(tonumber(t[2])/1000)})); return fresh`,
39	claim: `
40	redis.pcall('XGROUP','CREATE','bg:jobs','workers','0','MKSTREAM')
41	local id=redis.call('SPOP','bg:reclaim')
42	local rows
43	if id then rows=redis.call('XCLAIM','bg:jobs','workers',ARGV[1],0,id)
44	else local read=redis.call('XREADGROUP','GROUP','workers',ARGV[1],'COUNT',1,'STREAMS','bg:jobs','>'); if read then rows=read[1][2] end end
45	if not rows or #rows==0 then return nil end
46	local row=rows[1]; local fields=row[2]; local job=fields[2]; local kind=fields[4]
47	local t=redis.call('TIME'); local at=tonumber(t[1])*1000+math.floor(tonumber(t[2])/1000)
48	redis.call('RPUSH','bg:events',cjson.encode({event=id and 'reclaimed' or 'claimed',machine=ARGV[1],version=ARGV[2],job=job,id=row[1],at=at}))
49	return cjson.encode({id=row[1],job=job,kind=kind,checkpoint=redis.call('HGET','bg:checkpoints',job) or ''})`,
50	finish: `
51	local fresh=redis.call('HSETNX','bg:results',ARGV[2],ARGV[3])
52	local ack=redis.call('XACK','bg:jobs','workers',ARGV[1])
53	local t=redis.call('TIME'); local at=tonumber(t[1])*1000+math.floor(tonumber(t[2])/1000)
54	redis.call('RPUSH','bg:events',cjson.encode({event='acked',machine=ARGV[3],job=ARGV[2],id=ARGV[1],fresh=fresh,ack=ack,at=at})); return ack`,
55	checkpoint: `
56	redis.call('HSET','bg:checkpoints',ARGV[2],'saved-step-1')
57	redis.call('SADD','bg:reclaim',ARGV[1])
58	local t=redis.call('TIME'); local at=tonumber(t[1])*1000+math.floor(tonumber(t[2])/1000)
59	redis.call('RPUSH','bg:events',cjson.encode({event='checkpoint',machine=ARGV[3],job=ARGV[2],id=ARGV[1],at=at})); return 1`,
60	snapshot: `
61	local pending=redis.pcall('XPENDING','bg:jobs','workers')
62	return cjson.encode({events=redis.call('LRANGE','bg:events',0,-1),results=redis.call('HGETALL','bg:results'),checkpoints=redis.call('HGETALL','bg:checkpoints'),produced=redis.call('SMEMBERS','bg:produced'),entries=redis.call('XLEN','bg:jobs'),pending=pending.err and 0 or pending[1]})`,
63	} as const;
64
65	export interface Job {
66	id: string;
67	job: string;
68	kind: string;
69	checkpoint: string;
70	}
71	export interface LedgerEvent {
72	event: string;
73	machine: string;
74	version?: string;
75	job?: string;
76	id?: string;
77	at: number;
78	ack?: number;
79	fresh?: number;
80	worker?: string;
81	signal?: string;
82	}
83	export interface Snapshot {
84	events: string[];
85	results: string[];
86	checkpoints: string[];
87	produced: string[];
88	pending: number;
89	entries: number;
90	}