Monorepo for wisp.place. A static site hosting service built on top of the AT Protocol.
wisp.place
nekomimi.pet
1import { logger } from './observability';
2
3export type UploadJobStatus = 'pending' | 'processing' | 'uploading' | 'completed' | 'failed';
4
5export interface UploadProgress {
6 filesProcessed: number;
7 totalFiles: number;
8 filesUploaded: number;
9 filesReused: number;
10 currentFile?: string;
11 phase: 'validating' | 'compressing' | 'uploading' | 'creating_manifest' | 'finalizing' | 'done';
12}
13
14export interface UploadJob {
15 id: string;
16 did: string;
17 siteName: string;
18 status: UploadJobStatus;
19 progress: UploadProgress;
20 result?: {
21 success: boolean;
22 uri?: string;
23 cid?: string;
24 fileCount?: number;
25 siteName?: string;
26 skippedFiles?: Array<{ name: string; reason: string }>;
27 uploadedCount?: number;
28 };
29 error?: string;
30 createdAt: number;
31 updatedAt: number;
32}
33
34// In-memory job storage
35const jobs = new Map<string, UploadJob>();
36
37// SSE connections for each job
38const jobListeners = new Map<string, Set<(event: string, data: any) => void>>();
39
40// Cleanup old jobs after 1 hour
41const JOB_TTL = 60 * 60 * 1000;
42
43export function createUploadJob(did: string, siteName: string, totalFiles: number): string {
44 const id = crypto.randomUUID();
45 const now = Date.now();
46
47 const job: UploadJob = {
48 id,
49 did,
50 siteName,
51 status: 'pending',
52 progress: {
53 filesProcessed: 0,
54 totalFiles,
55 filesUploaded: 0,
56 filesReused: 0,
57 phase: 'validating'
58 },
59 createdAt: now,
60 updatedAt: now
61 };
62
63 jobs.set(id, job);
64 logger.info(`Upload job created: ${id} for ${did}/${siteName} (${totalFiles} files)`);
65
66 // Schedule cleanup
67 setTimeout(() => {
68 jobs.delete(id);
69 jobListeners.delete(id);
70 logger.info(`Upload job cleaned up: ${id}`);
71 }, JOB_TTL);
72
73 return id;
74}
75
76export function getUploadJob(id: string): UploadJob | undefined {
77 return jobs.get(id);
78}
79
80export function updateUploadJob(
81 id: string,
82 updates: Partial<Omit<UploadJob, 'id' | 'did' | 'siteName' | 'createdAt'>>
83): void {
84 const job = jobs.get(id);
85 if (!job) {
86 logger.warn(`Attempted to update non-existent job: ${id}`);
87 return;
88 }
89
90 Object.assign(job, updates, { updatedAt: Date.now() });
91 jobs.set(id, job);
92
93 // Notify all listeners
94 const listeners = jobListeners.get(id);
95 if (listeners && listeners.size > 0) {
96 const eventData = {
97 status: job.status,
98 progress: job.progress,
99 result: job.result,
100 error: job.error
101 };
102
103 const failedListeners: Array<(event: string, data: any) => void> = [];
104 listeners.forEach(listener => {
105 try {
106 listener('progress', eventData);
107 } catch (err) {
108 // Client disconnected, remove this listener
109 failedListeners.push(listener);
110 }
111 });
112
113 // Remove failed listeners
114 failedListeners.forEach(listener => listeners.delete(listener));
115 }
116}
117
118export function completeUploadJob(id: string, result: UploadJob['result']): void {
119 updateUploadJob(id, {
120 status: 'completed',
121 progress: {
122 ...getUploadJob(id)!.progress,
123 phase: 'done'
124 },
125 result
126 });
127
128 // Send final event and close connections
129 setTimeout(() => {
130 const listeners = jobListeners.get(id);
131 if (listeners) {
132 listeners.forEach(listener => {
133 try {
134 listener('done', result);
135 } catch (err) {
136 // Client already disconnected, ignore
137 }
138 });
139 jobListeners.delete(id);
140 }
141 }, 100);
142}
143
144export function failUploadJob(id: string, error: string): void {
145 updateUploadJob(id, {
146 status: 'failed',
147 error
148 });
149
150 // Send error event and close connections
151 setTimeout(() => {
152 const listeners = jobListeners.get(id);
153 if (listeners) {
154 listeners.forEach(listener => {
155 try {
156 listener('error', { error });
157 } catch (err) {
158 // Client already disconnected, ignore
159 }
160 });
161 jobListeners.delete(id);
162 }
163 }, 100);
164}
165
166export function addJobListener(jobId: string, listener: (event: string, data: any) => void): () => void {
167 if (!jobListeners.has(jobId)) {
168 jobListeners.set(jobId, new Set());
169 }
170 jobListeners.get(jobId)!.add(listener);
171
172 // Return cleanup function
173 return () => {
174 const listeners = jobListeners.get(jobId);
175 if (listeners) {
176 listeners.delete(listener);
177 if (listeners.size === 0) {
178 jobListeners.delete(jobId);
179 }
180 }
181 };
182}
183
184export function updateJobProgress(
185 jobId: string,
186 progressUpdate: Partial<UploadProgress>
187): void {
188 const job = getUploadJob(jobId);
189 if (!job) return;
190
191 updateUploadJob(jobId, {
192 progress: {
193 ...job.progress,
194 ...progressUpdate
195 }
196 });
197}