From bcffbf838fd94727085e544c4311290b1fcef161 Mon Sep 17 00:00:00 2001 From: Sundaram Kumar Jha Date: Tue, 1 Sep 2026 19:16:11 +0000 Subject: [PATCH 1/7] feat(rpc): add LinkService deep module for link cache and persistence invariants --- packages/rpc/src/services/link-service.ts | 809 ++++++++++++++++++++++ 1 file changed, 809 insertions(+) create mode 100644 packages/rpc/src/services/link-service.ts diff --git a/packages/rpc/src/services/link-service.ts b/packages/rpc/src/services/link-service.ts new file mode 100644 index 000000000..6c226bdb4 --- /dev/null +++ b/packages/rpc/src/services/link-service.ts @@ -0,0 +1,809 @@ +import { and, eq, isNull, isUniqueViolationFor } from "@databuddy/db"; +import { linkFolders, links } from "@databuddy/db/schema"; +import { + abandonCachedLinkMutation as redisAbandon, + beginCachedLinkMutation as redisBegin, + type CachedLink, + type CachedLinkMutationNext, + finishCachedLinkMutation as redisFinish, + invalidateAgentContextSnapshotsForOwner, + setCachedLinkIfAbsent as redisSetIfAbsent, +} from "@databuddy/redis"; +import { isDeepLinkTarget } from "@databuddy/shared/constants/deep-link-apps"; +import { getErrorLogFields } from "@databuddy/shared/evlog-fields"; +import { randomUUIDv7 } from "bun"; +import { customAlphabet } from "nanoid"; +import { rpcError } from "../errors"; +import { logger } from "../lib/logger"; +import type { Context } from "../orpc"; + +type LinkRow = typeof links.$inferSelect; +type CacheableLink = Pick< + LinkRow, + | "id" + | "targetUrl" + | "expiresAt" + | "expiredRedirectUrl" + | "ogTitle" + | "ogDescription" + | "ogImageUrl" + | "ogVideoUrl" + | "iosUrl" + | "androidUrl" + | "deepLinkApp" +>; + +interface LinkCacheMutation { + id: string; + organizationId: string; + slug: string; + token: string; +} +type LinkCacheMutationRequest = Pick< + LinkCacheMutation, + "id" | "organizationId" | "slug" +> & { mode: "existing" | "new" }; + +function hasPostgresSqlState(error: unknown): boolean { + const seen = new Set(); + let current: unknown = error; + while (typeof current === "object" && current !== null) { + if (seen.has(current as object)) { + return false; + } + seen.add(current as object); + if ( + "code" in (current as Record) && + typeof (current as Record).code === "string" && + ((current as Record).code as string).length === 5 && + "severity" in (current as Record) && + typeof (current as Record).severity === "string" + ) { + return true; + } + current = + "cause" in (current as Record) + ? (current as Record).cause + : null; + } + return false; +} + +function validateDeepLinkConfiguration( + deepLinkApp: string | null | undefined, + targetUrl: string +): void { + if (!deepLinkApp) { + return; + } + if (isDeepLinkTarget(deepLinkApp, targetUrl)) { + return; + } + throw rpcError.badRequest( + "Deep link URLs must use HTTPS and match the selected app" + ); +} + +export function normalizeNullableText( + value: string | null | undefined +): string | null { + if (value == null) { + return null; + } + const trimmed = value.trim(); + return trimmed || null; +} + +export function normalizeTargetDomain( + value: string | null | undefined +): string | null { + const trimmed = normalizeNullableText(value); + if (!trimmed) { + return null; + } + try { + return new URL( + trimmed.includes("://") ? trimmed : `https://${trimmed}` + ).hostname.toLowerCase(); + } catch { + return trimmed.split("/")[0]?.toLowerCase() || null; + } +} + +function getTargetDomain(targetUrl: string): string | null { + try { + return new URL(targetUrl).hostname.toLowerCase(); + } catch { + return null; + } +} + +async function validateFolderId( + db: Context["db"], + folderId: string | null | undefined, + organizationId: string +): Promise { + const normalizedFolderId = folderId?.trim() || null; + if (!normalizedFolderId) { + return null; + } + const existing = await db + .select({ id: linkFolders.id }) + .from(linkFolders) + .where( + and( + eq(linkFolders.id, normalizedFolderId), + eq(linkFolders.organizationId, organizationId), + isNull(linkFolders.deletedAt) + ) + ) + .limit(1); + if (existing.length > 0) { + return normalizedFolderId; + } + throw rpcError.badRequest("Link folder does not exist in this organization"); +} + +function toCachedLink(link: CacheableLink): CachedLink { + return { + id: link.id, + targetUrl: link.targetUrl, + expiresAt: link.expiresAt?.toISOString() ?? null, + expiredRedirectUrl: link.expiredRedirectUrl, + ogTitle: link.ogTitle, + ogDescription: link.ogDescription, + ogImageUrl: link.ogImageUrl, + ogVideoUrl: link.ogVideoUrl, + iosUrl: link.iosUrl, + androidUrl: link.androidUrl, + deepLinkApp: link.deepLinkApp, + }; +} + +const generateLinkSlug = customAlphabet( + "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz", + 8 +); + +export interface LinkServiceDeps { + db: Context["db"]; +} + +export class LinkService { + private db: Context["db"]; + + constructor(deps: LinkServiceDeps) { + this.db = deps.db; + } + + // — cache helpers (locality: all lease handling co-located) — + private async abandonLinkCacheMutations( + mutations: LinkCacheMutation[], + reason: string + ): Promise { + await Promise.all( + mutations.map(async ({ id, organizationId, slug, token }) => { + try { + if (await redisAbandon(slug, token)) { + return; + } + logger.warn( + { linkId: id, organizationId, slug }, + "Lost link cache mutation lease while abandoning mutation" + ); + } catch (error) { + logger.error( + { + linkId: id, + organizationId, + slug, + reason, + ...getErrorLogFields(error), + }, + "Failed to abandon link cache mutation" + ); + } + }) + ); + } + + private async finishLinkCacheMutation( + mutation: LinkCacheMutation, + next: CachedLinkMutationNext, + reason: string + ): Promise { + const { id, organizationId, slug, token } = mutation; + try { + if (await redisFinish(slug, token, next)) { + return true; + } + logger.warn( + { linkId: id, organizationId, slug, reason }, + "Lost link cache mutation lease before cache finalization" + ); + } catch (error) { + logger.error( + { + linkId: id, + organizationId, + slug, + reason, + ...getErrorLogFields(error), + }, + "Failed to finalize link cache mutation" + ); + } + try { + await redisAbandon(slug, token); + } catch (error) { + logger.error( + { + linkId: id, + organizationId, + slug, + reason, + ...getErrorLogFields(error), + }, + "Failed to release link cache mutation after finalization failure" + ); + } + return false; + } + + private async backfillLinkCache( + slug: string, + link: CacheableLink & { organizationId: string }, + reason: string + ): Promise { + try { + if (await redisSetIfAbsent(slug, toCachedLink(link))) { + return; + } + logger.warn( + { linkId: link.id, organizationId: link.organizationId, slug, reason }, + "Link cache backfill did not replace an existing entry" + ); + } catch (error) { + logger.error( + { + linkId: link.id, + organizationId: link.organizationId, + slug, + reason, + ...getErrorLogFields(error), + }, + "Failed to backfill link cache" + ); + } + } + + private async tombstoneLinkCacheMutations( + mutations: LinkCacheMutation[], + reason: string + ): Promise { + await Promise.all( + mutations.map((m) => + this.finishLinkCacheMutation( + m, + { id: m.id, state: "tombstone" }, + reason + ) + ) + ); + } + + private async beginLinkCacheMutations( + requests: LinkCacheMutationRequest[] + ): Promise { + const mutations: LinkCacheMutation[] = []; + try { + for (const request of [...requests].sort((a, b) => + a.slug.localeCompare(b.slug) + )) { + const started = await redisBegin(request.slug, request); + if (started.state !== "acquired") { + await this.abandonLinkCacheMutations( + mutations, + "another mutation already owns this slug" + ); + return null; + } + mutations.push({ + id: request.id, + organizationId: request.organizationId, + slug: request.slug, + token: started.token, + }); + } + return mutations; + } catch (error) { + await this.abandonLinkCacheMutations( + mutations, + "failed before the database mutation started" + ); + throw error; + } + } + + private invalidateLinkAgentContext(organizationId: string): void { + invalidateAgentContextSnapshotsForOwner(organizationId).catch((error) => { + logger.error( + { organizationId, ...getErrorLogFields(error) }, + "Unexpected link agent-context invalidation failure" + ); + }); + } + + async getLinkOrThrow(id: string): Promise { + const [link] = await this.db + .select() + .from(links) + .where(and(eq(links.id, id), isNull(links.deletedAt))) + .limit(1); + if (!link) { + throw rpcError.notFound("link", id); + } + return link; + } + + async create(input: { + organizationId: string; + createdBy: string; + name: string; + targetUrl: string; + slug?: string; + folderId?: string | null; + expiresAt?: string | Date | null; + expiredRedirectUrl?: string | null; + ogTitle?: string | null; + ogDescription?: string | null; + ogImageUrl?: string | null; + ogVideoUrl?: string | null; + iosUrl?: string | null; + androidUrl?: string | null; + externalId?: string | null; + sourceType?: string | null; + sourceId?: string | null; + sourceOwnerId?: string | null; + targetDomain?: string | null; + deepLinkApp?: string | null; + }): Promise { + validateDeepLinkConfiguration(input.deepLinkApp, input.targetUrl); + const resolvedFolderId = await validateFolderId( + this.db, + input.folderId, + input.organizationId + ); + const targetDomain = + normalizeTargetDomain(input.targetDomain) ?? + getTargetDomain(input.targetUrl); + + const slugsToTry = input.slug + ? [input.slug] + : Array.from({ length: 10 }, () => generateLinkSlug()); + + for (const slug of slugsToTry) { + const linkId = randomUUIDv7(); + let cacheMutations: LinkCacheMutation[] = []; + if (input.slug) { + let started: LinkCacheMutation[] | null; + try { + started = await this.beginLinkCacheMutations([ + { + id: linkId, + mode: "new", + organizationId: input.organizationId, + slug, + }, + ]); + } catch (error) { + logger.error( + { slug, linkId, ...getErrorLogFields(error) }, + "Failed to begin link cache mutation before create" + ); + throw rpcError.serviceUnavailable( + 1, + "Link cache is temporarily unavailable; retry this custom slug" + ); + } + if (!started) { + throw rpcError.conflict( + "This slug is already taken or is being updated" + ); + } + cacheMutations = started; + } + const [cacheMutation] = cacheMutations; + let finalizedFailure = false; + try { + const [newLink] = await this.db + .insert(links) + .values({ + id: linkId, + slug, + organizationId: input.organizationId, + createdBy: input.createdBy, + folderId: resolvedFolderId, + name: input.name, + targetUrl: input.targetUrl, + targetDomain, + sourceType: normalizeNullableText(input.sourceType), + sourceId: normalizeNullableText(input.sourceId), + sourceOwnerId: normalizeNullableText(input.sourceOwnerId), + expiresAt: input.expiresAt ? new Date(input.expiresAt) : null, + expiredRedirectUrl: input.expiredRedirectUrl ?? null, + ogTitle: input.ogTitle ?? null, + ogDescription: input.ogDescription ?? null, + ogImageUrl: input.ogImageUrl ?? null, + ogVideoUrl: input.ogVideoUrl ?? null, + iosUrl: input.iosUrl ?? null, + androidUrl: input.androidUrl ?? null, + externalId: input.externalId ?? null, + deepLinkApp: input.deepLinkApp ?? null, + }) + .returning(); + if (!newLink) { + await this.abandonLinkCacheMutations( + cacheMutations, + "create returned no persisted link" + ); + finalizedFailure = true; + throw rpcError.internal("Failed to create link"); + } + const publishCache = cacheMutation + ? this.finishLinkCacheMutation( + cacheMutation, + { link: toCachedLink(newLink), state: "link" }, + "create persisted" + ) + : this.backfillLinkCache( + slug, + newLink as CacheableLink & { organizationId: string }, + "create bypassed cache lease" + ); + publishCache.catch((error) => { + logger.error( + { slug, linkId, ...getErrorLogFields(error) }, + "Failed to publish created link to cache" + ); + }); + this.invalidateLinkAgentContext(input.organizationId); + return newLink; + } catch (error) { + if (isUniqueViolationFor(error, "links_slug_unique")) { + await this.abandonLinkCacheMutations( + cacheMutations, + "create failed because the slug already exists" + ); + if (input.slug) { + throw rpcError.conflict("This slug is already taken"); + } + continue; + } + if (finalizedFailure) { + throw error; + } + if (hasPostgresSqlState(error)) { + await this.abandonLinkCacheMutations( + cacheMutations, + "create failed with a definitive PostgreSQL error" + ); + throw error; + } + let persistedLink: LinkRow | undefined; + try { + [persistedLink] = await this.db + .select() + .from(links) + .where(eq(links.id, linkId)) + .limit(1); + } catch (reconciliationError) { + logger.error( + { slug, linkId, ...getErrorLogFields(reconciliationError) }, + "Failed to reconcile uncertain link create" + ); + throw rpcError.serviceUnavailable( + 1, + "Link creation outcome is still being reconciled" + ); + } + if (persistedLink) { + if (cacheMutation) { + await this.finishLinkCacheMutation( + cacheMutation, + { link: toCachedLink(persistedLink), state: "link" }, + "create reconciled after ambiguous database error" + ); + } else { + await this.backfillLinkCache( + slug, + persistedLink as CacheableLink & { organizationId: string }, + "create reconciled after cache bypass" + ); + } + this.invalidateLinkAgentContext(input.organizationId); + return persistedLink; + } + logger.error( + { slug, linkId, ...getErrorLogFields(error) }, + "Link create failed with an uncertain persistence outcome" + ); + throw rpcError.serviceUnavailable( + 1, + "Link creation outcome is still being reconciled" + ); + } + } + throw rpcError.internal("Failed to generate unique slug"); + } + + async update( + link: LinkRow, + input: { + name?: string; + targetUrl?: string; + slug?: string; + folderId?: string | null; + expiresAt?: string | Date | null; + expiredRedirectUrl?: string | null; + ogTitle?: string | null; + ogDescription?: string | null; + ogImageUrl?: string | null; + ogVideoUrl?: string | null; + iosUrl?: string | null; + androidUrl?: string | null; + externalId?: string | null; + sourceType?: string | null; + sourceId?: string | null; + sourceOwnerId?: string | null; + targetDomain?: string | null; + deepLinkApp?: string | null; + } + ): Promise { + if (input.targetUrl !== undefined || input.deepLinkApp !== undefined) { + validateDeepLinkConfiguration( + input.deepLinkApp === undefined ? link.deepLinkApp : input.deepLinkApp, + input.targetUrl ?? link.targetUrl + ); + } + let resolvedFolderId: string | null | undefined; + if (input.folderId !== undefined) { + resolvedFolderId = await validateFolderId( + this.db, + input.folderId, + link.organizationId + ); + } + const { + expiresAt, + folderId, + sourceType, + sourceId, + sourceOwnerId, + targetDomain, + ...updates + } = input as Record & typeof input; + const oldSlug = link.slug; + const nextSlug = (updates as { slug?: string }).slug ?? oldSlug; + const nextTargetDomain = + targetDomain === undefined + ? input.targetUrl + ? getTargetDomain(input.targetUrl) + : undefined + : normalizeTargetDomain(targetDomain); + + const cacheMutationRequests: LinkCacheMutationRequest[] = [ + { + id: link.id, + mode: "existing", + organizationId: link.organizationId, + slug: oldSlug, + }, + ]; + if (nextSlug !== oldSlug) { + cacheMutationRequests.push({ + id: link.id, + mode: "new", + organizationId: link.organizationId, + slug: nextSlug, + }); + } + let cacheMutations: LinkCacheMutation[] | null; + try { + cacheMutations = await this.beginLinkCacheMutations( + cacheMutationRequests + ); + } catch (error) { + logger.error( + { slug: oldSlug, linkId: link.id, ...getErrorLogFields(error) }, + "Failed to begin link cache mutation before update" + ); + throw rpcError.internal("Failed to update cache. Link not updated."); + } + if (!cacheMutations) { + logger.warn( + { linkId: link.id, oldSlug, nextSlug }, + "Link cache mutation conflicts with an in-progress or stale cache entry" + ); + throw rpcError.conflict( + "This link is currently being updated. Retry the request." + ); + } + const oldCacheMutation = cacheMutations.find((m) => m.slug === oldSlug); + if (!oldCacheMutation) { + await this.abandonLinkCacheMutations( + cacheMutations, + "missing old-slug cache mutation before update" + ); + throw rpcError.internal("Failed to begin link cache mutation"); + } + let finalizedFailure = false; + try { + const [updatedLink] = await this.db + .update(links) + .set({ + ...(updates as Record), + folderId: + resolvedFolderId === undefined ? undefined : resolvedFolderId, + sourceType: + sourceType === undefined + ? undefined + : normalizeNullableText(sourceType as string | null), + sourceId: + sourceId === undefined + ? undefined + : normalizeNullableText(sourceId as string | null), + sourceOwnerId: + sourceOwnerId === undefined + ? undefined + : normalizeNullableText(sourceOwnerId as string | null), + targetDomain: nextTargetDomain, + expiresAt: + expiresAt === undefined + ? undefined + : expiresAt + ? new Date(expiresAt as string) + : null, + updatedAt: new Date(), + }) + .where(eq(links.id, link.id)) + .returning(); + if (!updatedLink) { + await this.tombstoneLinkCacheMutations( + cacheMutations, + "update did not return a persisted link" + ); + finalizedFailure = true; + throw rpcError.notFound("link", link.id); + } + const cachedLink = toCachedLink(updatedLink); + if (nextSlug === oldSlug) { + await this.finishLinkCacheMutation( + oldCacheMutation, + { link: cachedLink, state: "link" }, + "update persisted" + ); + } else { + const newCacheMutation = cacheMutations.find( + (m) => m.slug === nextSlug + ); + if (newCacheMutation) { + await Promise.all([ + this.finishLinkCacheMutation( + oldCacheMutation, + { id: link.id, state: "tombstone" }, + "update renamed link" + ), + this.finishLinkCacheMutation( + newCacheMutation, + { link: cachedLink, state: "link" }, + "update renamed link" + ), + ]); + } else { + await this.tombstoneLinkCacheMutations( + cacheMutations, + "update persisted without a new-slug cache mutation" + ); + } + } + this.invalidateLinkAgentContext(link.organizationId); + return updatedLink; + } catch (error) { + if (isUniqueViolationFor(error, "links_slug_unique")) { + await this.abandonLinkCacheMutations( + cacheMutations, + "update failed because the new slug already exists" + ); + throw rpcError.conflict("This slug is already taken"); + } + if (finalizedFailure) { + throw error; + } + if (hasPostgresSqlState(error)) { + await this.abandonLinkCacheMutations( + cacheMutations, + "update failed with a definitive PostgreSQL error" + ); + throw error; + } + logger.error( + { linkId: link.id, oldSlug, nextSlug, ...getErrorLogFields(error) }, + "Link update failed with an uncertain persistence outcome" + ); + throw rpcError.serviceUnavailable( + 1, + "Link update outcome is still being reconciled" + ); + } + } + + async delete(link: LinkRow): Promise<{ success: true }> { + let cacheMutations: LinkCacheMutation[] | null; + try { + cacheMutations = await this.beginLinkCacheMutations([ + { + id: link.id, + mode: "existing", + organizationId: link.organizationId, + slug: link.slug, + }, + ]); + } catch (error) { + logger.error( + { slug: link.slug, linkId: link.id, ...getErrorLogFields(error) }, + "Failed to begin link cache mutation before delete" + ); + throw rpcError.internal("Failed to update cache. Link not deleted."); + } + if (!cacheMutations) { + logger.warn( + { slug: link.slug, linkId: link.id }, + "Link cache mutation conflicts with an in-progress or stale cache entry" + ); + throw rpcError.conflict( + "This link is currently being updated. Retry the request." + ); + } + let finalizedFailure = false; + try { + const deleted = await this.db + .delete(links) + .where(eq(links.id, link.id)) + .returning({ id: links.id }); + if (deleted.length === 0) { + await this.tombstoneLinkCacheMutations( + cacheMutations, + "delete did not remove a persisted link" + ); + finalizedFailure = true; + throw rpcError.notFound("link", link.id); + } + await this.tombstoneLinkCacheMutations( + cacheMutations, + "delete persisted" + ); + } catch (error) { + if (finalizedFailure) { + throw error; + } + if (hasPostgresSqlState(error)) { + await this.abandonLinkCacheMutations( + cacheMutations, + "delete failed with a definitive PostgreSQL error" + ); + throw error; + } + logger.error( + { slug: link.slug, linkId: link.id, ...getErrorLogFields(error) }, + "Link delete failed with an uncertain persistence outcome" + ); + throw rpcError.serviceUnavailable( + 1, + "Link deletion outcome is still being reconciled" + ); + } + this.invalidateLinkAgentContext(link.organizationId); + return { success: true }; + } +} From 304d416b0a5c14c13d22e826a5ecb0d060b8905f Mon Sep 17 00:00:00 2001 From: Sundaram Kumar Jha Date: Tue, 1 Sep 2026 19:16:16 +0000 Subject: [PATCH 2/7] refactor(rpc): thin links router to delegate to LinkService --- packages/rpc/src/routers/links.ts | 819 ++---------------------------- 1 file changed, 40 insertions(+), 779 deletions(-) diff --git a/packages/rpc/src/routers/links.ts b/packages/rpc/src/routers/links.ts index 0464f5a15..a18d85b41 100644 --- a/packages/rpc/src/routers/links.ts +++ b/packages/rpc/src/routers/links.ts @@ -7,26 +7,10 @@ import { ilike, isNotNull, isNull, - isUniqueViolationFor, or, } from "@databuddy/db"; -import { linkFolders, links } from "@databuddy/db/schema"; -import { - abandonCachedLinkMutation, - beginCachedLinkMutation, - type CachedLink, - type CachedLinkMutationNext, - finishCachedLinkMutation, - invalidateAgentContextSnapshotsForOwner, - setCachedLinkIfAbsent, -} from "@databuddy/redis"; -import { isDeepLinkTarget } from "@databuddy/shared/constants/deep-link-apps"; -import { randomUUIDv7 } from "bun"; -import { customAlphabet } from "nanoid"; +import { links } from "@databuddy/db/schema"; import { z } from "zod"; -import { rpcError } from "../errors"; -import { getErrorLogFields } from "@databuddy/shared/evlog-fields"; -import { logger } from "../lib/logger"; import { setTrackProperties } from "../middleware/track-mutation"; import { type Context, protectedProcedure, trackedProcedure } from "../orpc"; import { requireLinkAccess, requireOrganizationId } from "./link-access"; @@ -40,315 +24,9 @@ import { listLinksSchema, updateLinkSchema, } from "./links.schemas"; +import { LinkService, normalizeTargetDomain } from "../services/link-service"; -type LinkRow = typeof links.$inferSelect; -type CacheableLink = Pick< - LinkRow, - | "id" - | "targetUrl" - | "expiresAt" - | "expiredRedirectUrl" - | "ogTitle" - | "ogDescription" - | "ogImageUrl" - | "ogVideoUrl" - | "iosUrl" - | "androidUrl" - | "deepLinkApp" ->; -interface LinkCacheMutation { - id: string; - organizationId: string; - slug: string; - token: string; -} -type LinkCacheMutationRequest = Pick< - LinkCacheMutation, - "id" | "organizationId" | "slug" -> & { - mode: "existing" | "new"; -}; - -const generateLinkSlug = customAlphabet( - "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz", - 8 -); -function hasPostgresSqlState(error: unknown): boolean { - const seen = new Set(); - let current = error; - - while (typeof current === "object" && current !== null) { - if (seen.has(current)) { - return false; - } - seen.add(current); - - if ( - "code" in current && - typeof current.code === "string" && - current.code.length === 5 && - "severity" in current && - typeof current.severity === "string" - ) { - return true; - } - - current = "cause" in current ? current.cause : null; - } - - return false; -} - -function validateDeepLinkConfiguration( - deepLinkApp: string | null | undefined, - targetUrl: string -): void { - if (!deepLinkApp) { - return; - } - if (isDeepLinkTarget(deepLinkApp, targetUrl)) { - return; - } - throw rpcError.badRequest( - "Deep link URLs must use HTTPS and match the selected app" - ); -} - -function normalizeNullableText( - value: string | null | undefined -): string | null { - if (value == null) { - return null; - } - const trimmed = value.trim(); - return trimmed || null; -} - -function normalizeTargetDomain( - value: string | null | undefined -): string | null { - const trimmed = normalizeNullableText(value); - if (!trimmed) { - return null; - } - - try { - return new URL( - trimmed.includes("://") ? trimmed : `https://${trimmed}` - ).hostname.toLowerCase(); - } catch { - return trimmed.split("/")[0]?.toLowerCase() || null; - } -} - -function getTargetDomain(targetUrl: string): string | null { - try { - return new URL(targetUrl).hostname.toLowerCase(); - } catch { - return null; - } -} - -async function validateFolderId( - db: Context["db"], - folderId: string | null | undefined, - organizationId: string -): Promise { - const normalizedFolderId = folderId?.trim() || null; - if (!normalizedFolderId) { - return null; - } - - const existing = await db - .select({ id: linkFolders.id }) - .from(linkFolders) - .where( - and( - eq(linkFolders.id, normalizedFolderId), - eq(linkFolders.organizationId, organizationId), - isNull(linkFolders.deletedAt) - ) - ) - .limit(1); - - if (existing.length > 0) { - return normalizedFolderId; - } - - throw rpcError.badRequest("Link folder does not exist in this organization"); -} - -function toCachedLink(link: CacheableLink): CachedLink { - return { - id: link.id, - targetUrl: link.targetUrl, - expiresAt: link.expiresAt?.toISOString() ?? null, - expiredRedirectUrl: link.expiredRedirectUrl, - ogTitle: link.ogTitle, - ogDescription: link.ogDescription, - ogImageUrl: link.ogImageUrl, - ogVideoUrl: link.ogVideoUrl, - iosUrl: link.iosUrl, - androidUrl: link.androidUrl, - deepLinkApp: link.deepLinkApp, - }; -} - -function invalidateLinkAgentContext(organizationId: string): void { - // The helper absorbs expected Redis failures; retain this guard for future errors. - invalidateAgentContextSnapshotsForOwner(organizationId).catch((error) => { - logger.error( - { organizationId, ...getErrorLogFields(error) }, - "Unexpected link agent-context invalidation failure" - ); - }); -} - -async function abandonLinkCacheMutations( - mutations: LinkCacheMutation[], - reason: string -): Promise { - await Promise.all( - mutations.map(async ({ id, organizationId, slug, token }) => { - try { - if (await abandonCachedLinkMutation(slug, token)) { - return; - } - logger.warn( - { linkId: id, organizationId, slug }, - "Lost link cache mutation lease while abandoning mutation" - ); - } catch (error) { - logger.error( - { - linkId: id, - organizationId, - slug, - reason, - ...getErrorLogFields(error), - }, - "Failed to abandon link cache mutation" - ); - } - }) - ); -} - -async function finishLinkCacheMutation( - mutation: LinkCacheMutation, - next: CachedLinkMutationNext, - reason: string -): Promise { - const { id, organizationId, slug, token } = mutation; - try { - if (await finishCachedLinkMutation(slug, token, next)) { - return true; - } - logger.warn( - { linkId: id, organizationId, slug, reason }, - "Lost link cache mutation lease before cache finalization" - ); - } catch (error) { - logger.error( - { linkId: id, organizationId, slug, reason, ...getErrorLogFields(error) }, - "Failed to finalize link cache mutation" - ); - } - - // The database outcome is confirmed before callers finalize a mutation. Clear - // only this token's pending marker so redirects can read through to Postgres; - // a newer cache owner or already-applied finalization is never overwritten. - try { - await abandonCachedLinkMutation(slug, token); - } catch (error) { - logger.error( - { linkId: id, organizationId, slug, reason, ...getErrorLogFields(error) }, - "Failed to release link cache mutation after finalization failure" - ); - } - return false; -} - -async function backfillLinkCache( - slug: string, - link: CacheableLink & { organizationId: string }, - reason: string -): Promise { - try { - if (await setCachedLinkIfAbsent(slug, toCachedLink(link))) { - return; - } - logger.warn( - { - linkId: link.id, - organizationId: link.organizationId, - slug, - reason, - }, - "Link cache backfill did not replace an existing entry" - ); - } catch (error) { - logger.error( - { - linkId: link.id, - organizationId: link.organizationId, - slug, - reason, - ...getErrorLogFields(error), - }, - "Failed to backfill link cache" - ); - } -} - -async function tombstoneLinkCacheMutations( - mutations: LinkCacheMutation[], - reason: string -): Promise { - await Promise.all( - mutations.map((mutation) => - finishLinkCacheMutation( - mutation, - { id: mutation.id, state: "tombstone" }, - reason - ) - ) - ); -} - -async function beginLinkCacheMutations( - requests: LinkCacheMutationRequest[] -): Promise { - const mutations: LinkCacheMutation[] = []; - try { - for (const request of [...requests].sort((left, right) => - left.slug.localeCompare(right.slug) - )) { - const started = await beginCachedLinkMutation(request.slug, request); - if (started.state !== "acquired") { - await abandonLinkCacheMutations( - mutations, - "another mutation already owns this slug" - ); - return null; - } - mutations.push({ - id: request.id, - organizationId: request.organizationId, - slug: request.slug, - token: started.token, - }); - } - return mutations; - } catch (error) { - await abandonLinkCacheMutations( - mutations, - "failed before the database mutation started" - ); - throw error; - } -} - +// — query helpers stay in the router: they are shallow filters, not cache invariants. const LINKS_LIST_MAX = 1000; const ILIKE_PATTERN_CHARACTER_REGEX = /[\\%_]/g; @@ -398,7 +76,6 @@ function buildLinkSearchCondition(search: string | undefined) { if (!trimmed) { return; } - const term = `%${trimmed.replace(ILIKE_PATTERN_CHARACTER_REGEX, "\\$&")}%`; return or( ilike(links.name, term), @@ -419,20 +96,23 @@ const linkSortOrder = { "name-desc": [desc(links.name), desc(links.id)], } as const; -async function getLinkOrThrow(context: Context, id: string): Promise { +async function getLinkOrThrow(context: Context, id: string) { const [link] = await context.db .select() .from(links) .where(and(eq(links.id, id), isNull(links.deletedAt))) .limit(1); - if (!link) { + const { rpcError } = await import("../errors"); throw rpcError.notFound("link", id); } - return link; } +function createLinkService(context: Context): LinkService { + return new LinkService({ db: context.db }); +} + export const linksRouter = { list: protectedProcedure .route({ @@ -450,11 +130,8 @@ export const linksRouter = { const organizationId = requireOrganizationId( input.organizationId ?? context.organizationId ); - await requireLinkAccess(context, organizationId, "read"); - const conditions = buildLinkListConditions(input, organizationId); - return context.db .select() .from(links) @@ -479,22 +156,17 @@ export const linksRouter = { const organizationId = requireOrganizationId( input.organizationId ?? context.organizationId ); - await requireLinkAccess(context, organizationId, "read"); - const conditions = buildLinkListConditions(input, organizationId); - if (input.type === "short") { conditions.push(isNull(links.deepLinkApp)); } else if (input.type === "deep") { conditions.push(isNotNull(links.deepLinkApp)); } - const matches = buildLinkSearchCondition(input.search); if (matches) { conditions.push(matches); } - const where = and(...conditions); const pageQuery = context.db .select() @@ -503,9 +175,8 @@ export const linksRouter = { .orderBy(...linkSortOrder[input.sort]) .limit(input.limit + 1) .offset(input.offset); - let rows: LinkRow[]; + let rows: (typeof links.$inferSelect)[]; let total: number | undefined; - if (input.includeTotal) { const [page, [summary]] = await Promise.all([ pageQuery, @@ -516,9 +187,7 @@ export const linksRouter = { } else { rows = await pageQuery; } - const hasMore = rows.length > input.limit; - return { items: hasMore ? rows.slice(0, input.limit) : rows, hasMore, @@ -541,7 +210,6 @@ export const linksRouter = { .handler(async ({ context, input }) => { const link = await getLinkOrThrow(context, input.id); await requireLinkAccess(context, link.organizationId, "read"); - return link; }), @@ -564,190 +232,35 @@ export const linksRouter = { const organizationId = requireOrganizationId( input.organizationId?.trim() || context.organizationId ); - const workspace = await requireLinkAccess( context, organizationId, "create" ); - - validateDeepLinkConfiguration(input.deepLinkApp, input.targetUrl); - const [createdBy, resolvedFolderId] = await Promise.all([ - workspace.getCreatedBy(), - validateFolderId(context.db, input.folderId, organizationId), - ]); - const targetDomain = - normalizeTargetDomain(input.targetDomain) ?? - getTargetDomain(input.targetUrl); - - const slugsToTry = input.slug - ? [input.slug] - : Array.from({ length: 10 }, () => generateLinkSlug()); - - for (const slug of slugsToTry) { - const linkId = randomUUIDv7(); - // Custom slugs take a cache lease so hot-slug readers never see - // stale state. Generated slugs skip it: PostgreSQL's unique - // constraint is authoritative, nobody can read a slug before this - // response returns it, and redirects read through to PG on misses. - let cacheMutations: LinkCacheMutation[] = []; - if (input.slug) { - let started: LinkCacheMutation[] | null; - try { - started = await beginLinkCacheMutations([ - { id: linkId, mode: "new", organizationId, slug }, - ]); - } catch (error) { - logger.error( - { slug, linkId, ...getErrorLogFields(error) }, - "Failed to begin link cache mutation before create" - ); - throw rpcError.serviceUnavailable( - 1, - "Link cache is temporarily unavailable; retry this custom slug" - ); - } - if (!started) { - throw rpcError.conflict( - "This slug is already taken or is being updated" - ); - } - cacheMutations = started; - } - - const [cacheMutation] = cacheMutations; - - let finalizedFailure = false; - try { - const [newLink] = await context.db - .insert(links) - .values({ - id: linkId, - slug, - organizationId, - createdBy, - folderId: resolvedFolderId, - name: input.name, - targetUrl: input.targetUrl, - targetDomain, - sourceType: normalizeNullableText(input.sourceType), - sourceId: normalizeNullableText(input.sourceId), - sourceOwnerId: normalizeNullableText(input.sourceOwnerId), - expiresAt: input.expiresAt ? new Date(input.expiresAt) : null, - expiredRedirectUrl: input.expiredRedirectUrl ?? null, - ogTitle: input.ogTitle ?? null, - ogDescription: input.ogDescription ?? null, - ogImageUrl: input.ogImageUrl ?? null, - ogVideoUrl: input.ogVideoUrl ?? null, - iosUrl: input.iosUrl ?? null, - androidUrl: input.androidUrl ?? null, - externalId: input.externalId ?? null, - deepLinkApp: input.deepLinkApp ?? null, - }) - .returning(); - - if (!newLink) { - await abandonLinkCacheMutations( - cacheMutations, - "create returned no persisted link" - ); - finalizedFailure = true; - throw rpcError.internal("Failed to create link"); - } - - const publishCache = cacheMutation - ? finishLinkCacheMutation( - cacheMutation, - { link: toCachedLink(newLink), state: "link" }, - "create persisted" - ) - : backfillLinkCache(slug, newLink, "create bypassed cache lease"); - publishCache.catch((error) => { - logger.error( - { slug, linkId, ...getErrorLogFields(error) }, - "Failed to publish created link to cache" - ); - }); - invalidateLinkAgentContext(organizationId); - - return newLink; - } catch (error) { - if (isUniqueViolationFor(error, "links_slug_unique")) { - await abandonLinkCacheMutations( - cacheMutations, - "create failed because the slug already exists" - ); - if (input.slug) { - throw rpcError.conflict("This slug is already taken"); - } - continue; - } - if (finalizedFailure) { - throw error; - } - if (hasPostgresSqlState(error)) { - await abandonLinkCacheMutations( - cacheMutations, - "create failed with a definitive PostgreSQL error" - ); - throw error; - } - - let persistedLink: LinkRow | undefined; - try { - [persistedLink] = await context.db - .select() - .from(links) - .where(eq(links.id, linkId)) - .limit(1); - } catch (reconciliationError) { - logger.error( - { - slug, - linkId, - ...getErrorLogFields(reconciliationError), - }, - "Failed to reconcile uncertain link create" - ); - throw rpcError.serviceUnavailable( - 1, - "Link creation outcome is still being reconciled" - ); - } - - if (persistedLink) { - if (cacheMutation) { - await finishLinkCacheMutation( - cacheMutation, - { link: toCachedLink(persistedLink), state: "link" }, - "create reconciled after ambiguous database error" - ); - } else { - await backfillLinkCache( - slug, - persistedLink, - "create reconciled after cache bypass" - ); - } - invalidateLinkAgentContext(organizationId); - return persistedLink; - } - - logger.error( - { slug, linkId, ...getErrorLogFields(error) }, - "Link create failed with an uncertain persistence outcome" - ); - // A separate read that does not see the row cannot prove an in-flight - // COMMIT rolled back. Keep the short lease so no negative read-through - // can outlive a late commit, and tell the caller to retry. - throw rpcError.serviceUnavailable( - 1, - "Link creation outcome is still being reconciled" - ); - } - } - - throw rpcError.internal("Failed to generate unique slug"); + const createdBy = await workspace.getCreatedBy(); + const service = createLinkService(context); + return service.create({ + organizationId, + createdBy, + name: input.name, + targetUrl: input.targetUrl, + slug: input.slug, + folderId: input.folderId, + expiresAt: input.expiresAt ?? null, + expiredRedirectUrl: input.expiredRedirectUrl ?? null, + ogTitle: input.ogTitle ?? null, + ogDescription: input.ogDescription ?? null, + ogImageUrl: input.ogImageUrl ?? null, + ogVideoUrl: input.ogVideoUrl ?? null, + iosUrl: input.iosUrl ?? null, + androidUrl: input.androidUrl ?? null, + externalId: input.externalId ?? null, + sourceType: input.sourceType ?? null, + sourceId: input.sourceId ?? null, + sourceOwnerId: input.sourceOwnerId ?? null, + targetDomain: input.targetDomain ?? null, + deepLinkApp: input.deepLinkApp ?? null, + }); }), update: trackedProcedure @@ -764,196 +277,10 @@ export const linksRouter = { .handler(async ({ context, input }) => { const link = await getLinkOrThrow(context, input.id); await requireLinkAccess(context, link.organizationId, "update"); - - if (input.targetUrl !== undefined || input.deepLinkApp !== undefined) { - validateDeepLinkConfiguration( - input.deepLinkApp === undefined - ? link.deepLinkApp - : input.deepLinkApp, - input.targetUrl ?? link.targetUrl - ); - } - - let resolvedFolderId: string | null | undefined; - if (input.folderId !== undefined) { - resolvedFolderId = await validateFolderId( - context.db, - input.folderId, - link.organizationId - ); - } - - const { - id, - expiresAt, - folderId, - sourceType, - sourceId, - sourceOwnerId, - targetDomain, - ...updates - } = input; - const oldSlug = link.slug; - const nextSlug = updates.slug ?? oldSlug; - const nextTargetDomain = - targetDomain === undefined - ? input.targetUrl - ? getTargetDomain(input.targetUrl) - : undefined - : normalizeTargetDomain(targetDomain); - - const cacheMutationRequests: LinkCacheMutationRequest[] = [ - { - id: link.id, - mode: "existing", - organizationId: link.organizationId, - slug: oldSlug, - }, - ]; - if (nextSlug !== oldSlug) { - cacheMutationRequests.push({ - id: link.id, - mode: "new", - organizationId: link.organizationId, - slug: nextSlug, - }); - } - - let cacheMutations: LinkCacheMutation[] | null; - try { - cacheMutations = await beginLinkCacheMutations(cacheMutationRequests); - } catch (error) { - logger.error( - { slug: oldSlug, linkId: link.id, ...getErrorLogFields(error) }, - "Failed to begin link cache mutation before update" - ); - throw rpcError.internal("Failed to update cache. Link not updated."); - } - - if (!cacheMutations) { - logger.warn( - { linkId: link.id, oldSlug, nextSlug }, - "Link cache mutation conflicts with an in-progress or stale cache entry" - ); - throw rpcError.conflict( - "This link is currently being updated. Retry the request." - ); - } - - const oldCacheMutation = cacheMutations.find( - (mutation) => mutation.slug === oldSlug - ); - if (!oldCacheMutation) { - await abandonLinkCacheMutations( - cacheMutations, - "missing old-slug cache mutation before update" - ); - throw rpcError.internal("Failed to begin link cache mutation"); - } - - let finalizedFailure = false; - try { - const [updatedLink] = await context.db - .update(links) - .set({ - ...updates, - folderId: - resolvedFolderId === undefined ? undefined : resolvedFolderId, - sourceType: - sourceType === undefined - ? undefined - : normalizeNullableText(sourceType), - sourceId: - sourceId === undefined - ? undefined - : normalizeNullableText(sourceId), - sourceOwnerId: - sourceOwnerId === undefined - ? undefined - : normalizeNullableText(sourceOwnerId), - targetDomain: nextTargetDomain, - expiresAt: - expiresAt === undefined - ? undefined - : expiresAt - ? new Date(expiresAt) - : null, - updatedAt: new Date(), - }) - .where(eq(links.id, id)) - .returning(); - - if (!updatedLink) { - await tombstoneLinkCacheMutations( - cacheMutations, - "update did not return a persisted link" - ); - finalizedFailure = true; - throw rpcError.notFound("link", input.id); - } - - const cachedLink = toCachedLink(updatedLink); - if (nextSlug === oldSlug) { - await finishLinkCacheMutation( - oldCacheMutation, - { link: cachedLink, state: "link" }, - "update persisted" - ); - } else { - const newCacheMutation = cacheMutations.find( - (mutation) => mutation.slug === nextSlug - ); - if (newCacheMutation) { - await Promise.all([ - finishLinkCacheMutation( - oldCacheMutation, - { id: link.id, state: "tombstone" }, - "update renamed link" - ), - finishLinkCacheMutation( - newCacheMutation, - { link: cachedLink, state: "link" }, - "update renamed link" - ), - ]); - } else { - await tombstoneLinkCacheMutations( - cacheMutations, - "update persisted without a new-slug cache mutation" - ); - } - } - - invalidateLinkAgentContext(link.organizationId); - - return updatedLink; - } catch (error) { - if (isUniqueViolationFor(error, "links_slug_unique")) { - await abandonLinkCacheMutations( - cacheMutations, - "update failed because the new slug already exists" - ); - throw rpcError.conflict("This slug is already taken"); - } - if (finalizedFailure) { - throw error; - } - if (hasPostgresSqlState(error)) { - await abandonLinkCacheMutations( - cacheMutations, - "update failed with a definitive PostgreSQL error" - ); - throw error; - } - logger.error( - { linkId: link.id, oldSlug, nextSlug, ...getErrorLogFields(error) }, - "Link update failed with an uncertain persistence outcome" - ); - throw rpcError.serviceUnavailable( - 1, - "Link update outcome is still being reconciled" - ); - } + const service = createLinkService(context); + // Pass only updatable fields; id is used to fetch the row. + const { id: _id, ...updates } = input; + return service.update(link, updates); }), delete: trackedProcedure @@ -970,73 +297,7 @@ export const linksRouter = { .handler(async ({ context, input }) => { const link = await getLinkOrThrow(context, input.id); await requireLinkAccess(context, link.organizationId, "delete"); - - let cacheMutations: LinkCacheMutation[] | null; - try { - cacheMutations = await beginLinkCacheMutations([ - { - id: link.id, - mode: "existing", - organizationId: link.organizationId, - slug: link.slug, - }, - ]); - } catch (error) { - logger.error( - { slug: link.slug, linkId: input.id, ...getErrorLogFields(error) }, - "Failed to begin link cache mutation before delete" - ); - throw rpcError.internal("Failed to update cache. Link not deleted."); - } - - if (!cacheMutations) { - logger.warn( - { slug: link.slug, linkId: link.id }, - "Link cache mutation conflicts with an in-progress or stale cache entry" - ); - throw rpcError.conflict( - "This link is currently being updated. Retry the request." - ); - } - - let finalizedFailure = false; - try { - const deleted = await context.db - .delete(links) - .where(eq(links.id, input.id)) - .returning({ id: links.id }); - if (deleted.length === 0) { - await tombstoneLinkCacheMutations( - cacheMutations, - "delete did not remove a persisted link" - ); - finalizedFailure = true; - throw rpcError.notFound("link", input.id); - } - - await tombstoneLinkCacheMutations(cacheMutations, "delete persisted"); - } catch (error) { - if (finalizedFailure) { - throw error; - } - if (hasPostgresSqlState(error)) { - await abandonLinkCacheMutations( - cacheMutations, - "delete failed with a definitive PostgreSQL error" - ); - throw error; - } - logger.error( - { slug: link.slug, linkId: link.id, ...getErrorLogFields(error) }, - "Link delete failed with an uncertain persistence outcome" - ); - throw rpcError.serviceUnavailable( - 1, - "Link deletion outcome is still being reconciled" - ); - } - invalidateLinkAgentContext(link.organizationId); - - return { success: true }; + const service = createLinkService(context); + return service.delete(link); }), }; From 1c8f507571c39007509873b584f935c9c0d11a35 Mon Sep 17 00:00:00 2001 From: Sundaram Kumar Jha Date: Tue, 1 Sep 2026 19:16:19 +0000 Subject: [PATCH 3/7] test(rpc): add LinkService tests for helpers and cache lease invariants --- .../rpc/src/services/link-service.test.ts | 315 ++++++++++++++++++ 1 file changed, 315 insertions(+) create mode 100644 packages/rpc/src/services/link-service.test.ts diff --git a/packages/rpc/src/services/link-service.test.ts b/packages/rpc/src/services/link-service.test.ts new file mode 100644 index 000000000..0e881d5dc --- /dev/null +++ b/packages/rpc/src/services/link-service.test.ts @@ -0,0 +1,315 @@ +import { afterAll, beforeEach, describe, expect, it, mock } from "bun:test"; + +import { normalizeNullableText, normalizeTargetDomain } from "./link-service"; + +// — pure helpers (no mocks needed) — +describe("normalizeNullableText", () => { + it("trims and nulls empty strings", () => { + expect(normalizeNullableText(null)).toBeNull(); + expect(normalizeNullableText(undefined)).toBeNull(); + expect(normalizeNullableText("")).toBeNull(); + expect(normalizeNullableText(" ")).toBeNull(); + expect(normalizeNullableText(" hello ")).toBe("hello"); + expect(normalizeNullableText("a")).toBe("a"); + }); +}); + +describe("normalizeTargetDomain", () => { + it("extracts hostname and lowercases", () => { + expect(normalizeTargetDomain("https://Example.COM/path")).toBe( + "example.com" + ); + expect(normalizeTargetDomain("example.com")).toBe("example.com"); + expect(normalizeTargetDomain(" WWW.example.com/ ")).toBe( + "www.example.com" + ); + expect(normalizeTargetDomain(null)).toBeNull(); + expect(normalizeTargetDomain(" ")).toBeNull(); + expect(normalizeTargetDomain("not a url /")).toBe("not a url "); + }); +}); + +// — LinkService with mocked Redis + fake DB — +// Redis is imported directly by the service, so mock the module before importing the service. +const mockBegin = mock(async () => ({ state: "acquired", token: "tok-1" })); +const mockFinish = mock(async () => true); +const mockAbandon = mock(async () => true); +const mockSetIfAbsent = mock(async () => true); +const mockInvalidate = mock(async () => undefined); +const mockLoggerError = mock(() => undefined); +const mockLoggerWarn = mock(() => undefined); + +mock.module("@databuddy/redis", () => ({ + abandonCachedLinkMutation: mockAbandon, + beginCachedLinkMutation: mockBegin, + finishCachedLinkMutation: mockFinish, + invalidateAgentContextSnapshotsForOwner: mockInvalidate, + setCachedLinkIfAbsent: mockSetIfAbsent, +})); + +mock.module("../lib/logger", () => ({ + logger: { + error: mockLoggerError, + warn: mockLoggerWarn, + info: mock(() => undefined), + }, +})); + +const { LinkService } = await import("./link-service"); + +afterAll(() => { + mock.restore(); +}); + +type FakeLink = { + id: string; + slug: string; + organizationId: string; + targetUrl: string; + targetDomain: string | null; + name: string; + folderId: string | null; + deepLinkApp: string | null; + expiresAt: Date | null; + deletedAt: null; + createdAt: Date; + updatedAt: Date; +}; + +function makeFakeDb(opts: { + folderExists?: boolean; + insertImpl?: (values: Record) => Promise; + selectImpl?: () => Promise; + updateImpl?: () => Promise; + deleteImpl?: () => Promise<{ id: string }[]>; +} = {}) { + const folderExists = opts.folderExists ?? true; + const links = new Map(); + + return { + links, + db: { + select: (..._args: unknown[]) => ({ + from: (..._f: unknown[]) => ({ + where: (..._w: unknown[]) => ({ + limit: async (n: number) => { + if (opts.selectImpl) return opts.selectImpl(); + // folder check path — called with { id: linkFolders.id } + // distinguish by checking if we are in folder validation context: + // the service checks folder existence: return 1 row if folderExists + // heuristic: if select was called with { id: ... } shape, treat as folder + // we track via a flag — for simplicity return folder row when folderExists + // and not handling link reconciliation here + // To differentiate link reconciliation (select by id), we check mock state + // For now, if opts.selectImpl not provided, assume folder check + return folderExists ? [{ id: "folder-1" }].slice(0, n) : []; + }, + }), + }), + }), + insert: (..._a: unknown[]) => ({ + values: (values: Record) => ({ + returning: async () => { + if (opts.insertImpl) return opts.insertImpl(values); + const row: FakeLink = { + id: values.id as string, + slug: values.slug as string, + organizationId: values.organizationId as string, + targetUrl: values.targetUrl as string, + targetDomain: (values.targetDomain as string | null) ?? null, + name: values.name as string, + folderId: (values.folderId as string | null) ?? null, + deepLinkApp: (values.deepLinkApp as string | null) ?? null, + expiresAt: (values.expiresAt as Date | null) ?? null, + deletedAt: null, + createdAt: new Date(), + updatedAt: new Date(), + }; + links.set(row.id, row); + return [row]; + }, + }), + }), + update: (..._a: unknown[]) => ({ + set: (..._s: unknown[]) => ({ + where: (..._w: unknown[]) => ({ + returning: async () => { + if (opts.updateImpl) return opts.updateImpl(); + return []; + }, + }), + }), + }), + delete: (..._a: unknown[]) => ({ + where: (..._w: unknown[]) => ({ + returning: async () => { + if (opts.deleteImpl) return opts.deleteImpl(); + return []; + }, + }), + }), + } as unknown as ConstructorParameters[0]["db"], + }; +} + +beforeEach(() => { + mockBegin.mockClear(); + mockFinish.mockClear(); + mockAbandon.mockClear(); + mockSetIfAbsent.mockClear(); + mockInvalidate.mockClear(); + mockLoggerError.mockClear(); + mockLoggerWarn.mockClear(); + mockBegin.mockImplementation(async () => ({ state: "acquired", token: "tok-1" })); + mockFinish.mockImplementation(async () => true); + mockAbandon.mockImplementation(async () => true); + mockSetIfAbsent.mockImplementation(async () => true); + mockInvalidate.mockImplementation(async () => undefined); +}); + +describe("LinkService", () => { + it("rejects deep-link mismatches before touching DB", async () => { + const { db } = makeFakeDb(); + const svc = new LinkService({ db }); + + await expect( + svc.create({ + organizationId: "org-1", + createdBy: "user-1", + name: "Bad Deep", + targetUrl: "https://example.com/page", + deepLinkApp: "instagram", + }) + ).rejects.toMatchObject({ code: "BAD_REQUEST" }); + + expect(mockBegin).not.toHaveBeenCalled(); + }); + + it("rejects unknown folder", async () => { + const { db } = makeFakeDb({ folderExists: false }); + const svc = new LinkService({ db }); + + await expect( + svc.create({ + organizationId: "org-1", + createdBy: "user-1", + name: "With Folder", + targetUrl: "https://example.com", + folderId: "missing-folder", + }) + ).rejects.toMatchObject({ code: "BAD_REQUEST" }); + }); + + it("creates with custom slug via cache lease", async () => { + const { db, links } = makeFakeDb(); + const svc = new LinkService({ db }); + + const link = await svc.create({ + organizationId: "org-1", + createdBy: "user-1", + name: "My Link", + targetUrl: "https://example.com/a", + slug: "my-slug", + }); + + expect(link.slug).toBe("my-slug"); + expect(links.size).toBe(1); + expect(mockBegin).toHaveBeenCalledTimes(1); + expect(mockBegin.mock.calls[0]?.[0]).toBe("my-slug"); + }); + + it("throws conflict when cache lease is busy for custom slug", async () => { + mockBegin.mockImplementationOnce(async () => ({ state: "busy" })); + const { db } = makeFakeDb(); + const svc = new LinkService({ db }); + + await expect( + svc.create({ + organizationId: "org-1", + createdBy: "user-1", + name: "My Link", + targetUrl: "https://example.com", + slug: "taken-slug", + }) + ).rejects.toMatchObject({ code: "CONFLICT" }); + }); + + it("throws serviceUnavailable when cache is temporarily unavailable", async () => { + mockBegin.mockImplementationOnce(async () => { + throw new Error("redis down"); + }); + const { db } = makeFakeDb(); + const svc = new LinkService({ db }); + + await expect( + svc.create({ + organizationId: "org-1", + createdBy: "user-1", + name: "My Link", + targetUrl: "https://example.com", + slug: "any-slug", + }) + ).rejects.toMatchObject({ code: "SERVICE_UNAVAILABLE" }); + }); + + it("retries generated slugs on PG unique violation", async () => { + let call = 0; + const { db } = makeFakeDb({ + insertImpl: async (values) => { + call += 1; + if (call === 1) { + throw Object.assign(new Error("duplicate"), { + code: "23505", + constraint: "links_slug_unique", + severity: "ERROR", + }); + } + return [ + { + id: values.id as string, + slug: values.slug as string, + organizationId: values.organizationId as string, + targetUrl: values.targetUrl as string, + targetDomain: null, + name: values.name as string, + folderId: null, + deepLinkApp: null, + expiresAt: null, + deletedAt: null, + createdAt: new Date(), + updatedAt: new Date(), + } as FakeLink, + ]; + }, + }); + const svc = new LinkService({ db }); + + const link = await svc.create({ + organizationId: "org-1", + createdBy: "user-1", + name: "Generated", + targetUrl: "https://example.com", + }); + + expect(link.slug).toBeDefined(); + expect(call).toBe(2); + expect(mockBegin).not.toHaveBeenCalled(); + }); + + it("publishes via backfill when generated slug skips lease", async () => { + const { db } = makeFakeDb(); + const svc = new LinkService({ db }); + + await svc.create({ + organizationId: "org-1", + createdBy: "user-1", + name: "Gen", + targetUrl: "https://example.com", + }); + + // wait a tick for fire-and-forget backfill + await new Promise((r) => setTimeout(r, 10)); + expect(mockSetIfAbsent).toHaveBeenCalledTimes(1); + expect(mockFinish).not.toHaveBeenCalled(); + }); +}); From 250b5d13290a3e224a5ca767844e397d87de5001 Mon Sep 17 00:00:00 2001 From: Sundaram Kumar Jha Date: Tue, 1 Sep 2026 19:35:32 +0000 Subject: [PATCH 4/7] fix(rpc,redis): defer createdBy until validation, overwrite negative cache on backfill and abandon lease on ambiguous create --- packages/redis/links-cache.ts | 42 +++++++++++++++++++ packages/rpc/src/routers/links.ts | 3 +- .../rpc/src/services/link-service.test.ts | 6 ++- packages/rpc/src/services/link-service.ts | 33 +++++++++++++-- 4 files changed, 77 insertions(+), 7 deletions(-) diff --git a/packages/redis/links-cache.ts b/packages/redis/links-cache.ts index 4889dff4b..63847e5ce 100644 --- a/packages/redis/links-cache.ts +++ b/packages/redis/links-cache.ts @@ -321,6 +321,48 @@ export async function setCachedLinkIfAbsent( ); return result === "OK"; } + +const SET_CACHED_LINK_IF_ABSENT_OR_NOT_FOUND_SCRIPT = ` +-- set-cached-link-if-absent-or-not-found +local current = redis.call("GET", KEYS[1]) +if current == false then + redis.call("SET", KEYS[1], ARGV[1], "EX", ARGV[2]) + return 1 +end +if current == "null" then + redis.call("SET", KEYS[1], ARGV[1], "EX", ARGV[2]) + return 1 +end +local ok, decoded = pcall(cjson.decode, current) +if not ok or type(decoded) ~= "table" then + redis.call("SET", KEYS[1], ARGV[1], "EX", ARGV[2]) + return 1 +end +if decoded.state == "tombstone" then + redis.call("SET", KEYS[1], ARGV[1], "EX", ARGV[2]) + return 1 +end +if decoded.state == "pending" then + return 0 +end +return 0 +`; + +export async function setCachedLinkIfAbsentOrNotFound( + slug: string, + link: CachedLink +): Promise { + const result = (await runLinkCacheCommand((redis) => + redis.eval( + SET_CACHED_LINK_IF_ABSENT_OR_NOT_FOUND_SCRIPT, + 1, + getLinkCacheKey(slug), + JSON.stringify(link), + String(LINKS_CACHE_TTL) + ) + )) as number; + return result === 1; +} export async function setCachedLinkNotFoundIfAbsent( slug: string ): Promise { diff --git a/packages/rpc/src/routers/links.ts b/packages/rpc/src/routers/links.ts index a18d85b41..7256105dc 100644 --- a/packages/rpc/src/routers/links.ts +++ b/packages/rpc/src/routers/links.ts @@ -237,11 +237,10 @@ export const linksRouter = { organizationId, "create" ); - const createdBy = await workspace.getCreatedBy(); const service = createLinkService(context); return service.create({ organizationId, - createdBy, + getCreatedBy: () => workspace.getCreatedBy(), name: input.name, targetUrl: input.targetUrl, slug: input.slug, diff --git a/packages/rpc/src/services/link-service.test.ts b/packages/rpc/src/services/link-service.test.ts index 0e881d5dc..8bd99ba5a 100644 --- a/packages/rpc/src/services/link-service.test.ts +++ b/packages/rpc/src/services/link-service.test.ts @@ -35,6 +35,7 @@ const mockBegin = mock(async () => ({ state: "acquired", token: "tok-1" })); const mockFinish = mock(async () => true); const mockAbandon = mock(async () => true); const mockSetIfAbsent = mock(async () => true); +const mockSetIfAbsentOrNotFound = mock(async () => true); const mockInvalidate = mock(async () => undefined); const mockLoggerError = mock(() => undefined); const mockLoggerWarn = mock(() => undefined); @@ -45,6 +46,7 @@ mock.module("@databuddy/redis", () => ({ finishCachedLinkMutation: mockFinish, invalidateAgentContextSnapshotsForOwner: mockInvalidate, setCachedLinkIfAbsent: mockSetIfAbsent, + setCachedLinkIfAbsentOrNotFound: mockSetIfAbsentOrNotFound, })); mock.module("../lib/logger", () => ({ @@ -157,6 +159,7 @@ beforeEach(() => { mockFinish.mockClear(); mockAbandon.mockClear(); mockSetIfAbsent.mockClear(); + mockSetIfAbsentOrNotFound.mockClear(); mockInvalidate.mockClear(); mockLoggerError.mockClear(); mockLoggerWarn.mockClear(); @@ -164,6 +167,7 @@ beforeEach(() => { mockFinish.mockImplementation(async () => true); mockAbandon.mockImplementation(async () => true); mockSetIfAbsent.mockImplementation(async () => true); + mockSetIfAbsentOrNotFound.mockImplementation(async () => true); mockInvalidate.mockImplementation(async () => undefined); }); @@ -309,7 +313,7 @@ describe("LinkService", () => { // wait a tick for fire-and-forget backfill await new Promise((r) => setTimeout(r, 10)); - expect(mockSetIfAbsent).toHaveBeenCalledTimes(1); + expect(mockSetIfAbsentOrNotFound).toHaveBeenCalledTimes(1); expect(mockFinish).not.toHaveBeenCalled(); }); }); diff --git a/packages/rpc/src/services/link-service.ts b/packages/rpc/src/services/link-service.ts index 6c226bdb4..7be3e9f78 100644 --- a/packages/rpc/src/services/link-service.ts +++ b/packages/rpc/src/services/link-service.ts @@ -7,7 +7,7 @@ import { type CachedLinkMutationNext, finishCachedLinkMutation as redisFinish, invalidateAgentContextSnapshotsForOwner, - setCachedLinkIfAbsent as redisSetIfAbsent, + setCachedLinkIfAbsentOrNotFound as redisSetIfAbsentOrNotFound, } from "@databuddy/redis"; import { isDeepLinkTarget } from "@databuddy/shared/constants/deep-link-apps"; import { getErrorLogFields } from "@databuddy/shared/evlog-fields"; @@ -256,7 +256,7 @@ export class LinkService { reason: string ): Promise { try { - if (await redisSetIfAbsent(slug, toCachedLink(link))) { + if (await redisSetIfAbsentOrNotFound(slug, toCachedLink(link))) { return; } logger.warn( @@ -348,7 +348,8 @@ export class LinkService { async create(input: { organizationId: string; - createdBy: string; + createdBy?: string; + getCreatedBy?: () => Promise; name: string; targetUrl: string; slug?: string; @@ -377,6 +378,14 @@ export class LinkService { const targetDomain = normalizeTargetDomain(input.targetDomain) ?? getTargetDomain(input.targetUrl); + let createdBy: string; + if (input.getCreatedBy) { + createdBy = await input.getCreatedBy(); + } else if (input.createdBy) { + createdBy = input.createdBy; + } else { + throw new Error("createdBy or getCreatedBy required"); + } const slugsToTry = input.slug ? [input.slug] @@ -422,7 +431,7 @@ export class LinkService { id: linkId, slug, organizationId: input.organizationId, - createdBy: input.createdBy, + createdBy, folderId: resolvedFolderId, name: input.name, targetUrl: input.targetUrl, @@ -502,6 +511,10 @@ export class LinkService { { slug, linkId, ...getErrorLogFields(reconciliationError) }, "Failed to reconcile uncertain link create" ); + await this.abandonLinkCacheMutations( + cacheMutations, + "create reconciled with error" + ); throw rpcError.serviceUnavailable( 1, "Link creation outcome is still being reconciled" @@ -528,6 +541,10 @@ export class LinkService { { slug, linkId, ...getErrorLogFields(error) }, "Link create failed with an uncertain persistence outcome" ); + await this.abandonLinkCacheMutations( + cacheMutations, + "create failed with uncertain persistence outcome" + ); throw rpcError.serviceUnavailable( 1, "Link creation outcome is still being reconciled" @@ -731,6 +748,10 @@ export class LinkService { { linkId: link.id, oldSlug, nextSlug, ...getErrorLogFields(error) }, "Link update failed with an uncertain persistence outcome" ); + await this.abandonLinkCacheMutations( + cacheMutations, + "update failed with uncertain persistence outcome" + ); throw rpcError.serviceUnavailable( 1, "Link update outcome is still being reconciled" @@ -798,6 +819,10 @@ export class LinkService { { slug: link.slug, linkId: link.id, ...getErrorLogFields(error) }, "Link delete failed with an uncertain persistence outcome" ); + await this.abandonLinkCacheMutations( + cacheMutations, + "delete failed with uncertain persistence outcome" + ); throw rpcError.serviceUnavailable( 1, "Link deletion outcome is still being reconciled" From 9ca0cd41f3bd30711c82c75f165cc3c32d6be4b7 Mon Sep 17 00:00:00 2001 From: Sundaram Kumar Jha Date: Wed, 2 Sep 2026 01:17:19 +0530 Subject: [PATCH 5/7] refactor: update error message Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com> --- packages/rpc/src/services/link-service.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/rpc/src/services/link-service.ts b/packages/rpc/src/services/link-service.ts index 7be3e9f78..cdcc94a8e 100644 --- a/packages/rpc/src/services/link-service.ts +++ b/packages/rpc/src/services/link-service.ts @@ -384,7 +384,7 @@ export class LinkService { } else if (input.createdBy) { createdBy = input.createdBy; } else { - throw new Error("createdBy or getCreatedBy required"); +throw rpcError.internal("createdBy or getCreatedBy required"); } const slugsToTry = input.slug From 91e560ee8f6ed87a66972063980224643dfef9b4 Mon Sep 17 00:00:00 2001 From: Sundaram Kumar Jha Date: Wed, 2 Sep 2026 01:23:00 +0530 Subject: [PATCH 6/7] refactor: update indentation Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com> --- packages/rpc/src/services/link-service.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/rpc/src/services/link-service.ts b/packages/rpc/src/services/link-service.ts index cdcc94a8e..e58dd65da 100644 --- a/packages/rpc/src/services/link-service.ts +++ b/packages/rpc/src/services/link-service.ts @@ -384,7 +384,7 @@ export class LinkService { } else if (input.createdBy) { createdBy = input.createdBy; } else { -throw rpcError.internal("createdBy or getCreatedBy required"); + throw rpcError.internal("createdBy or getCreatedBy required"); } const slugsToTry = input.slug From f96f300de89f88411034ff995bb2a8153013ddb9 Mon Sep 17 00:00:00 2001 From: Sundaram Kumar Jha Date: Tue, 1 Sep 2026 20:06:58 +0000 Subject: [PATCH 7/7] fix(rpc,redis): retain lease on uncertain creates and prevent backfill from overwriting tombstones Keep pending lease for ambiguous inserts so late commits are not lost to negative caching or second writers; backfill now only replaces missing/null/corrupt entries, never pending or tombstone (prevents resurrecting deleted links). --- packages/redis/links-cache.ts | 3 +-- packages/rpc/src/services/link-service.ts | 16 ---------------- 2 files changed, 1 insertion(+), 18 deletions(-) diff --git a/packages/redis/links-cache.ts b/packages/redis/links-cache.ts index 63847e5ce..d6ecc528e 100644 --- a/packages/redis/links-cache.ts +++ b/packages/redis/links-cache.ts @@ -339,8 +339,7 @@ if not ok or type(decoded) ~= "table" then return 1 end if decoded.state == "tombstone" then - redis.call("SET", KEYS[1], ARGV[1], "EX", ARGV[2]) - return 1 + return 0 end if decoded.state == "pending" then return 0 diff --git a/packages/rpc/src/services/link-service.ts b/packages/rpc/src/services/link-service.ts index e58dd65da..274428fa9 100644 --- a/packages/rpc/src/services/link-service.ts +++ b/packages/rpc/src/services/link-service.ts @@ -511,10 +511,6 @@ export class LinkService { { slug, linkId, ...getErrorLogFields(reconciliationError) }, "Failed to reconcile uncertain link create" ); - await this.abandonLinkCacheMutations( - cacheMutations, - "create reconciled with error" - ); throw rpcError.serviceUnavailable( 1, "Link creation outcome is still being reconciled" @@ -541,10 +537,6 @@ export class LinkService { { slug, linkId, ...getErrorLogFields(error) }, "Link create failed with an uncertain persistence outcome" ); - await this.abandonLinkCacheMutations( - cacheMutations, - "create failed with uncertain persistence outcome" - ); throw rpcError.serviceUnavailable( 1, "Link creation outcome is still being reconciled" @@ -748,10 +740,6 @@ export class LinkService { { linkId: link.id, oldSlug, nextSlug, ...getErrorLogFields(error) }, "Link update failed with an uncertain persistence outcome" ); - await this.abandonLinkCacheMutations( - cacheMutations, - "update failed with uncertain persistence outcome" - ); throw rpcError.serviceUnavailable( 1, "Link update outcome is still being reconciled" @@ -819,10 +807,6 @@ export class LinkService { { slug: link.slug, linkId: link.id, ...getErrorLogFields(error) }, "Link delete failed with an uncertain persistence outcome" ); - await this.abandonLinkCacheMutations( - cacheMutations, - "delete failed with uncertain persistence outcome" - ); throw rpcError.serviceUnavailable( 1, "Link deletion outcome is still being reconciled"