mirror of
https://github.com/hanzo-apps/dataroom.git
synced 2026-10-30 14:47:36 +00:00
zeekayandHanzo Dev
6b68c9299e
The client was already @hanzo/kv; only the vocabulary was still Redis. Every module, symbol and comment now says what it connects to: lib/redis.ts -> lib/kv.ts (exports kv, lockKv) lib/redis-job-store.ts -> lib/kv-job-store.ts (KvJobStore) lib/redis-download-job-store.ts -> lib/kv-download-job-store.ts lib/files/tus-redis-locker.ts -> lib/files/tus-kv-locker.ts (KvLocker, KvLock) lib/api/domains/redis.ts -> lib/api/domains/kv.ts (getKey) lib/redis/ -> lib/kv-queue/ KV_URL is now the only endpoint variable — the REDIS_URL fallback and the redis://localhost default are gone, and the default scheme is kv://, which @hanzo/kv parses natively (utils/index.js). .env.example gained the KV_URL it never documented, and lost UPSTASH_REDIS_REST_LOCKER_URL/TOKEN: nothing has read those since the client changed — the locker opens a second connection to the same KV, so there is no second endpoint to configure. The second connection is `lockKv`, not `locker`: both tus routes already have a local `const locker = new KvLocker(...)`, so naming the client `locker` made the lock take itself as its own client. Verified against the repo's own gate (npx tsc --noEmit): 100 errors before and after, 0 introduced, compared set-wise by file:line:code. Behaviour is untouched — this is naming, one connection string, and dead config. Co-authored-by: Hanzo Dev <[email protected]>
122 lines
2.9 KiB
TypeScript
122 lines
2.9 KiB
TypeScript
import { kv } from "@/lib/kv";
|
|
|
|
const ITEM_TTL_SECONDS = 8 * 24 * 60 * 60; // 8 days
|
|
|
|
type QueueItem = {
|
|
dataroomDocumentId: string;
|
|
senderUserId: string;
|
|
queuedAt: number;
|
|
};
|
|
|
|
type DigestViewerEntry = {
|
|
viewerId: string;
|
|
dataroomId: string;
|
|
teamId: string;
|
|
};
|
|
|
|
function itemsKey(viewerId: string, dataroomId: string) {
|
|
return `dataroom_digest_items:${viewerId}:${dataroomId}`;
|
|
}
|
|
|
|
function viewerSetKey(frequency: "daily" | "weekly") {
|
|
return `dataroom_digest_viewers:${frequency}`;
|
|
}
|
|
|
|
function encodeViewerEntry(entry: DigestViewerEntry): string {
|
|
return `${entry.viewerId}:${entry.dataroomId}:${entry.teamId}`;
|
|
}
|
|
|
|
function decodeViewerEntry(encoded: string): DigestViewerEntry {
|
|
const [viewerId, dataroomId, teamId] = encoded.split(":");
|
|
return { viewerId, dataroomId, teamId };
|
|
}
|
|
|
|
export async function queueNotification({
|
|
frequency,
|
|
viewerId,
|
|
dataroomId,
|
|
teamId,
|
|
dataroomDocumentId,
|
|
senderUserId,
|
|
}: {
|
|
frequency: "daily" | "weekly";
|
|
viewerId: string;
|
|
dataroomId: string;
|
|
teamId: string;
|
|
dataroomDocumentId: string;
|
|
senderUserId: string;
|
|
}) {
|
|
const key = itemsKey(viewerId, dataroomId);
|
|
const item: QueueItem = {
|
|
dataroomDocumentId,
|
|
senderUserId,
|
|
queuedAt: Date.now(),
|
|
};
|
|
|
|
const pipeline = kv.pipeline();
|
|
pipeline.rpush(key, JSON.stringify(item));
|
|
pipeline.expire(key, ITEM_TTL_SECONDS);
|
|
pipeline.sadd(
|
|
viewerSetKey(frequency),
|
|
encodeViewerEntry({ viewerId, dataroomId, teamId }),
|
|
);
|
|
await pipeline.exec();
|
|
}
|
|
|
|
export type DigestBatch = {
|
|
viewerId: string;
|
|
dataroomId: string;
|
|
teamId: string;
|
|
items: QueueItem[];
|
|
};
|
|
|
|
export async function popDigestQueue(
|
|
frequency: "daily" | "weekly",
|
|
): Promise<DigestBatch[]> {
|
|
const setKey = viewerSetKey(frequency);
|
|
|
|
const members = await kv.smembers(setKey);
|
|
if (!members || members.length === 0) return [];
|
|
|
|
// Remove the entire set atomically
|
|
await kv.del(setKey);
|
|
|
|
const batches: DigestBatch[] = [];
|
|
|
|
for (const member of members) {
|
|
const entry = decodeViewerEntry(member);
|
|
const key = itemsKey(entry.viewerId, entry.dataroomId);
|
|
|
|
// Get all items then delete the list
|
|
const rawItems = await kv.lrange(key, 0, -1);
|
|
await kv.del(key);
|
|
|
|
if (!rawItems || rawItems.length === 0) continue;
|
|
|
|
const items: QueueItem[] = rawItems
|
|
.map((raw) => {
|
|
try {
|
|
return typeof raw === "string"
|
|
? (JSON.parse(raw) as QueueItem)
|
|
: (raw as QueueItem);
|
|
} catch (error) {
|
|
console.warn(
|
|
`[dataroom-digest] Skipping corrupted queue item for viewer=${entry.viewerId} dataroom=${entry.dataroomId}:`,
|
|
{ raw, error },
|
|
);
|
|
return null;
|
|
}
|
|
})
|
|
.filter((item): item is QueueItem => item !== null);
|
|
|
|
batches.push({
|
|
viewerId: entry.viewerId,
|
|
dataroomId: entry.dataroomId,
|
|
teamId: entry.teamId,
|
|
items,
|
|
});
|
|
}
|
|
|
|
return batches;
|
|
}
|