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
|
import { eq } from "drizzle-orm";
import { DequeuedJob, Runner } from "liteque";
import { db } from "@hoarder/db";
import { assets } from "@hoarder/db/schema";
import { deleteAsset, getAllAssets } from "@hoarder/shared/assetdb";
import logger from "@hoarder/shared/logger";
import {
TidyAssetsQueue,
ZTidyAssetsRequest,
zTidyAssetsRequestSchema,
} from "@hoarder/shared/queues";
export class TidyAssetsWorker {
static build() {
logger.info("Starting tidy assets worker ...");
const worker = new Runner<ZTidyAssetsRequest>(
TidyAssetsQueue,
{
run: runTidyAssets,
onComplete: (job) => {
const jobId = job.id;
logger.info(`[tidyAssets][${jobId}] Completed successfully`);
return Promise.resolve();
},
onError: (job) => {
const jobId = job.id;
logger.error(
`[tidyAssets][${jobId}] tidy assets job failed: ${job.error}\n${job.error.stack}`,
);
return Promise.resolve();
},
},
{
concurrency: 1,
pollIntervalMs: 1000,
timeoutSecs: 30,
},
);
return worker;
}
}
async function handleAsset(
asset: {
assetId: string;
userId: string;
size: number;
contentType: string;
fileName?: string | null;
},
request: ZTidyAssetsRequest,
jobId: string,
) {
const dbRow = await db.query.assets.findFirst({
where: eq(assets.id, asset.assetId),
});
if (!dbRow) {
if (request.cleanDanglingAssets) {
await deleteAsset({ userId: asset.userId, assetId: asset.assetId });
logger.info(
`[tidyAssets][${jobId}] Asset ${asset.assetId} not found in the database. Deleting it.`,
);
} else {
logger.warn(
`[tidyAssets][${jobId}] Asset ${asset.assetId} not found in the database. Not deleting it because cleanDanglingAssets is false.`,
);
}
return;
}
if (request.syncAssetMetadata) {
await db
.update(assets)
.set({
contentType: asset.contentType,
fileName: asset.fileName,
size: asset.size,
})
.where(eq(assets.id, asset.assetId));
logger.info(
`[tidyAssets][${jobId}] Updated metadata for asset ${asset.assetId}`,
);
}
}
async function runTidyAssets(job: DequeuedJob<ZTidyAssetsRequest>) {
const jobId = job.id;
const request = zTidyAssetsRequestSchema.safeParse(job.data);
if (!request.success) {
throw new Error(
`[tidyAssets][${jobId}] Got malformed job request: ${request.error.toString()}`,
);
}
for await (const asset of getAllAssets()) {
try {
handleAsset(asset, request.data, jobId);
} catch (e) {
logger.error(
`[tidyAssets][${jobId}] Failed to tidy asset ${asset.assetId}: ${e}`,
);
}
}
}
|
