Skip to content

Commit cc93284

Browse files
MohamedBassemMohamedBassem
committed
deps: upgrade liteque to 0.9.1 and move more transactions to begin immediate
1 parent 2f35474 commit cc93284

7 files changed

Lines changed: 201 additions & 186 deletions

File tree

‎apps/workers/package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@
3030
"https-proxy-agent": "^7.0.6",
3131
"ipaddr.js": "^2.2.0",
3232
"jsdom": "^24.0.0",
33-
"liteque": "^0.9.0",
33+
"liteque": "^0.9.1",
3434
"lru-cache": "^11.2.2",
3535
"metascraper": "^5.50.0",
3636
"metascraper-amazon": "^5.50.0",

‎apps/workers/workers/inference/tagging.ts‎

Lines changed: 83 additions & 78 deletions
Original file line numberDiff line numberDiff line change
@@ -420,90 +420,95 @@ async function connectTags(
420420
return;
421421
}
422422

423-
const res = await db.transaction((tx) => {
424-
// Attempt to match exiting tags with the new ones
425-
const { matchedTagIds, notFoundTagNames } = (() => {
426-
const { normalizeTag } = tagNormalizer();
427-
const normalizedInferredTags = inferredTags.map((t) => ({
428-
originalTag: t,
429-
normalizedTag: normalizeTag(t),
430-
}));
431-
432-
const matchedTags = tx.query.bookmarkTags
433-
.findMany({
434-
where: and(
435-
eq(bookmarkTags.userId, userId),
436-
inArray(
437-
bookmarkTags.normalizedName,
438-
normalizedInferredTags.map((t) => t.normalizedTag),
423+
// This transaction reads before writing, so reserve the writer slot before
424+
// taking a WAL snapshot that another connection could invalidate.
425+
const res = await db.transaction(
426+
(tx) => {
427+
// Attempt to match exiting tags with the new ones
428+
const { matchedTagIds, notFoundTagNames } = (() => {
429+
const { normalizeTag } = tagNormalizer();
430+
const normalizedInferredTags = inferredTags.map((t) => ({
431+
originalTag: t,
432+
normalizedTag: normalizeTag(t),
433+
}));
434+
435+
const matchedTags = tx.query.bookmarkTags
436+
.findMany({
437+
where: and(
438+
eq(bookmarkTags.userId, userId),
439+
inArray(
440+
bookmarkTags.normalizedName,
441+
normalizedInferredTags.map((t) => t.normalizedTag),
442+
),
439443
),
440-
),
441-
})
442-
.sync();
443-
444-
const matchedTagIds = matchedTags.map((r) => r.id);
445-
const notFoundTagNames = normalizedInferredTags
446-
.filter(
447-
(t) =>
448-
!matchedTags.some(
449-
(mt) => normalizeTag(mt.name) === t.normalizedTag,
450-
),
451-
)
452-
.map((t) => t.originalTag);
453-
454-
return { matchedTagIds, notFoundTagNames };
455-
})();
456-
457-
// Create tags that didn't exist previously
458-
let newTagIds: string[] = [];
459-
if (notFoundTagNames.length > 0) {
460-
newTagIds = tx
461-
.insert(bookmarkTags)
462-
.values(
463-
notFoundTagNames.map((t) => ({
464-
name: t,
465-
userId,
466-
})),
467-
)
468-
.onConflictDoNothing()
469-
.returning()
470-
.all()
471-
.map((t) => t.id);
472-
}
444+
})
445+
.sync();
446+
447+
const matchedTagIds = matchedTags.map((r) => r.id);
448+
const notFoundTagNames = normalizedInferredTags
449+
.filter(
450+
(t) =>
451+
!matchedTags.some(
452+
(mt) => normalizeTag(mt.name) === t.normalizedTag,
453+
),
454+
)
455+
.map((t) => t.originalTag);
456+
457+
return { matchedTagIds, notFoundTagNames };
458+
})();
459+
460+
// Create tags that didn't exist previously
461+
let newTagIds: string[] = [];
462+
if (notFoundTagNames.length > 0) {
463+
newTagIds = tx
464+
.insert(bookmarkTags)
465+
.values(
466+
notFoundTagNames.map((t) => ({
467+
name: t,
468+
userId,
469+
})),
470+
)
471+
.onConflictDoNothing()
472+
.returning()
473+
.all()
474+
.map((t) => t.id);
475+
}
473476

474-
// Delete old AI tags
475-
const detachedTags = tx
476-
.delete(tagsOnBookmarks)
477-
.where(
478-
and(
479-
eq(tagsOnBookmarks.attachedBy, "ai"),
480-
eq(tagsOnBookmarks.bookmarkId, bookmarkId),
481-
),
482-
)
483-
.returning()
484-
.all();
485-
486-
const allTagIds = new Set([...matchedTagIds, ...newTagIds]);
487-
488-
// Attach new ones
489-
let attachedTags: { tagId: string; bookmarkId: string }[] = [];
490-
if (allTagIds.size > 0) {
491-
attachedTags = tx
492-
.insert(tagsOnBookmarks)
493-
.values(
494-
[...allTagIds].map((tagId) => ({
495-
tagId,
496-
bookmarkId,
497-
attachedBy: "ai" as const,
498-
})),
477+
// Delete old AI tags
478+
const detachedTags = tx
479+
.delete(tagsOnBookmarks)
480+
.where(
481+
and(
482+
eq(tagsOnBookmarks.attachedBy, "ai"),
483+
eq(tagsOnBookmarks.bookmarkId, bookmarkId),
484+
),
499485
)
500-
.onConflictDoNothing()
501486
.returning()
502487
.all();
503-
}
504488

505-
return { detachedTags, attachedTags };
506-
});
489+
const allTagIds = new Set([...matchedTagIds, ...newTagIds]);
490+
491+
// Attach new ones
492+
let attachedTags: { tagId: string; bookmarkId: string }[] = [];
493+
if (allTagIds.size > 0) {
494+
attachedTags = tx
495+
.insert(tagsOnBookmarks)
496+
.values(
497+
[...allTagIds].map((tagId) => ({
498+
tagId,
499+
bookmarkId,
500+
attachedBy: "ai" as const,
501+
})),
502+
)
503+
.onConflictDoNothing()
504+
.returning()
505+
.all();
506+
}
507+
508+
return { detachedTags, attachedTags };
509+
},
510+
{ behavior: "immediate" },
511+
);
507512

508513
await RuleEngine.triggerOnEvent(userId, bookmarkId, [
509514
...res.detachedTags.map((t) => ({

‎packages/plugins/package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@
3131
"@restatedev/restate-sdk-clients": "^1.16.2",
3232
"async-mutex": "^0.4.1",
3333
"glob": "^11.0.0",
34-
"liteque": "^0.9.0",
34+
"liteque": "^0.9.1",
3535
"meilisearch": "^0.58.0",
3636
"redis": "^5.11.0"
3737
},

‎packages/trpc/models/importSessions.repo.ts‎

Lines changed: 64 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -110,68 +110,73 @@ export class ImportSessionsRepo {
110110

111111
// One transaction per session to keep write locks short; a large backlog
112112
// (e.g. the first sweep after deploy) shouldn't block other writers.
113+
// Each transaction reads before writing, so reserve the writer slot before
114+
// taking a WAL snapshot that another connection could invalidate.
113115
let archivedCount = 0;
114116
for (const session of sessions) {
115-
const archived = await this.db.transaction((tx) => {
116-
const statusCounts = tx
117-
.select({
118-
status: importStagingBookmarks.status,
119-
count: count(),
120-
})
121-
.from(importStagingBookmarks)
122-
.where(eq(importStagingBookmarks.importSessionId, session.id))
123-
.groupBy(importStagingBookmarks.status)
124-
.all();
125-
126-
const stats = {
127-
totalBookmarks: 0,
128-
completedBookmarks: 0,
129-
failedBookmarks: 0,
130-
pendingBookmarks: 0,
131-
processingBookmarks: 0,
132-
};
133-
134-
for (const { status, count: itemCount } of statusCounts) {
135-
stats.totalBookmarks += itemCount;
136-
switch (status) {
137-
case "pending":
138-
stats.pendingBookmarks += itemCount;
139-
break;
140-
case "processing":
141-
stats.processingBookmarks += itemCount;
142-
break;
143-
case "completed":
144-
stats.completedBookmarks += itemCount;
145-
break;
146-
case "failed":
147-
stats.failedBookmarks += itemCount;
148-
break;
117+
const archived = await this.db.transaction(
118+
(tx) => {
119+
const statusCounts = tx
120+
.select({
121+
status: importStagingBookmarks.status,
122+
count: count(),
123+
})
124+
.from(importStagingBookmarks)
125+
.where(eq(importStagingBookmarks.importSessionId, session.id))
126+
.groupBy(importStagingBookmarks.status)
127+
.all();
128+
129+
const stats = {
130+
totalBookmarks: 0,
131+
completedBookmarks: 0,
132+
failedBookmarks: 0,
133+
pendingBookmarks: 0,
134+
processingBookmarks: 0,
135+
};
136+
137+
for (const { status, count: itemCount } of statusCounts) {
138+
stats.totalBookmarks += itemCount;
139+
switch (status) {
140+
case "pending":
141+
stats.pendingBookmarks += itemCount;
142+
break;
143+
case "processing":
144+
stats.processingBookmarks += itemCount;
145+
break;
146+
case "completed":
147+
stats.completedBookmarks += itemCount;
148+
break;
149+
case "failed":
150+
stats.failedBookmarks += itemCount;
151+
break;
152+
}
149153
}
150-
}
151-
152-
const result = tx
153-
.update(importSessions)
154-
.set({ status: "archived", ...stats })
155-
.where(
156-
and(
157-
eq(importSessions.id, session.id),
158-
eq(importSessions.status, "completed"),
159-
),
160-
)
161-
.run();
162-
163-
if (result.changes === 0) {
164-
return false;
165-
}
166-
167-
tx.delete(importStagingBookmarks)
168-
.where(eq(importStagingBookmarks.importSessionId, session.id))
169-
.run();
170-
tx.delete(importSessionBookmarks)
171-
.where(eq(importSessionBookmarks.importSessionId, session.id))
172-
.run();
173-
return true;
174-
});
154+
155+
const result = tx
156+
.update(importSessions)
157+
.set({ status: "archived", ...stats })
158+
.where(
159+
and(
160+
eq(importSessions.id, session.id),
161+
eq(importSessions.status, "completed"),
162+
),
163+
)
164+
.run();
165+
166+
if (result.changes === 0) {
167+
return false;
168+
}
169+
170+
tx.delete(importStagingBookmarks)
171+
.where(eq(importStagingBookmarks.importSessionId, session.id))
172+
.run();
173+
tx.delete(importSessionBookmarks)
174+
.where(eq(importSessionBookmarks.importSessionId, session.id))
175+
.run();
176+
return true;
177+
},
178+
{ behavior: "immediate" },
179+
);
175180

176181
if (archived) {
177182
archivedCount++;

0 commit comments

Comments
 (0)