From 9a8bc060d90880209b094996fe1981bcd3e2cfda Mon Sep 17 00:00:00 2001 From: Richard R Date: Sat, 6 Jun 2026 19:12:40 -0600 Subject: [PATCH] feat(data): enhance anonymous claim to support document settings and TTS segment copy Expand the anonymous data claim process to include document settings transfer and TTS segment S3 prefix copying. Remove foreign key constraints from user_tts_chars to allow non-user buckets. Update claim modal and onboarding flow to display claimed document settings. Refactor TTS char count transfer to merge all dates and fix upsert logic. Add S3 copy utility for TTS segments and corresponding tests. Update migrations and schema to reflect relaxed constraints. --- .../0010_user-data-cleanup-cascades.sql | 7 +- drizzle/postgres/meta/0010_snapshot.json | 18 +- drizzle/postgres/meta/_journal.json | 2 +- .../0010_user-data-cleanup-cascades.sql | 19 +- drizzle/sqlite/meta/0010_snapshot.json | 18 +- drizzle/sqlite/meta/_journal.json | 2 +- src/app/api/user/claim/route.ts | 8 +- src/components/auth/ClaimDataModal.tsx | 6 +- src/contexts/OnboardingFlowContext.tsx | 8 +- src/db/schema_postgres.ts | 3 +- src/db/schema_sqlite.ts | 3 +- src/lib/server/auth/auth.ts | 140 ++--------- src/lib/server/rate-limit/rate-limiter.ts | 46 ++-- src/lib/server/tts/segments-blobstore.ts | 36 +++ src/lib/server/user/claim-data.ts | 223 ++++++++++++++++-- .../transfer-user-documents.vitest.spec.ts | 178 +++++++++++++- .../tts-segments-blobstore.vitest.spec.ts | 33 ++- 17 files changed, 518 insertions(+), 232 deletions(-) diff --git a/drizzle/postgres/0010_user-data-cleanup-cascades.sql b/drizzle/postgres/0010_user-data-cleanup-cascades.sql index 4040691..463adbd 100644 --- a/drizzle/postgres/0010_user-data-cleanup-cascades.sql +++ b/drizzle/postgres/0010_user-data-cleanup-cascades.sql @@ -1,10 +1,5 @@ ALTER TABLE "user_job_events" ADD CONSTRAINT "user_job_events_user_id_user_id_fk" FOREIGN KEY ("user_id") REFERENCES "public"."user"("id") ON DELETE cascade ON UPDATE no action NOT VALID;--> statement-breakpoint -ALTER TABLE "user_tts_chars" ADD CONSTRAINT "user_tts_chars_user_id_user_id_fk" FOREIGN KEY ("user_id") REFERENCES "public"."user"("id") ON DELETE cascade ON UPDATE no action NOT VALID;--> statement-breakpoint DELETE FROM "user_job_events" WHERE NOT EXISTS ( SELECT 1 FROM "user" WHERE "user"."id" = "user_job_events"."user_id" );--> statement-breakpoint -DELETE FROM "user_tts_chars" WHERE NOT EXISTS ( - SELECT 1 FROM "user" WHERE "user"."id" = "user_tts_chars"."user_id" -);--> statement-breakpoint -ALTER TABLE "user_job_events" VALIDATE CONSTRAINT "user_job_events_user_id_user_id_fk";--> statement-breakpoint -ALTER TABLE "user_tts_chars" VALIDATE CONSTRAINT "user_tts_chars_user_id_user_id_fk"; +ALTER TABLE "user_job_events" VALIDATE CONSTRAINT "user_job_events_user_id_user_id_fk"; \ No newline at end of file diff --git a/drizzle/postgres/meta/0010_snapshot.json b/drizzle/postgres/meta/0010_snapshot.json index 806434e..a85fbaa 100644 --- a/drizzle/postgres/meta/0010_snapshot.json +++ b/drizzle/postgres/meta/0010_snapshot.json @@ -1,5 +1,5 @@ { - "id": "7aaaea38-619e-42d8-b683-6231e6221d33", + "id": "1099cbe6-0c2e-4ea8-ae3c-6883c931169d", "prevId": "23b90a4b-fb63-4f9a-9230-09ca218d0b78", "version": "7", "dialect": "postgresql", @@ -1509,21 +1509,7 @@ "with": {} } }, - "foreignKeys": { - "user_tts_chars_user_id_user_id_fk": { - "name": "user_tts_chars_user_id_user_id_fk", - "tableFrom": "user_tts_chars", - "tableTo": "user", - "columnsFrom": [ - "user_id" - ], - "columnsTo": [ - "id" - ], - "onDelete": "cascade", - "onUpdate": "no action" - } - }, + "foreignKeys": {}, "compositePrimaryKeys": { "user_tts_chars_user_id_date_pk": { "name": "user_tts_chars_user_id_date_pk", diff --git a/drizzle/postgres/meta/_journal.json b/drizzle/postgres/meta/_journal.json index a93c07e..3429cfb 100644 --- a/drizzle/postgres/meta/_journal.json +++ b/drizzle/postgres/meta/_journal.json @@ -75,7 +75,7 @@ { "idx": 10, "version": "7", - "when": 1780785284335, + "when": 1780794520924, "tag": "0010_user-data-cleanup-cascades", "breakpoints": true } diff --git a/drizzle/sqlite/0010_user-data-cleanup-cascades.sql b/drizzle/sqlite/0010_user-data-cleanup-cascades.sql index b891cd9..be0e21c 100644 --- a/drizzle/sqlite/0010_user-data-cleanup-cascades.sql +++ b/drizzle/sqlite/0010_user-data-cleanup-cascades.sql @@ -2,9 +2,6 @@ PRAGMA foreign_keys=OFF;--> statement-breakpoint DELETE FROM `user_job_events` WHERE NOT EXISTS ( SELECT 1 FROM `user` WHERE `user`.`id` = `user_job_events`.`user_id` );--> statement-breakpoint -DELETE FROM `user_tts_chars` WHERE NOT EXISTS ( - SELECT 1 FROM `user` WHERE `user`.`id` = `user_tts_chars`.`user_id` -);--> statement-breakpoint CREATE TABLE `__new_user_job_events` ( `user_id` text NOT NULL, `action` text NOT NULL, @@ -18,18 +15,4 @@ INSERT INTO `__new_user_job_events`("user_id", "action", "op_id", "created_at") DROP TABLE `user_job_events`;--> statement-breakpoint ALTER TABLE `__new_user_job_events` RENAME TO `user_job_events`;--> statement-breakpoint PRAGMA foreign_keys=ON;--> statement-breakpoint -CREATE INDEX `idx_user_job_events_user_action_created` ON `user_job_events` (`user_id`,`action`,`created_at`);--> statement-breakpoint -CREATE TABLE `__new_user_tts_chars` ( - `user_id` text NOT NULL, - `date` text NOT NULL, - `char_count` integer DEFAULT 0, - `created_at` integer DEFAULT (cast(unixepoch('subsecond') * 1000 as integer)), - `updated_at` integer DEFAULT (cast(unixepoch('subsecond') * 1000 as integer)), - PRIMARY KEY(`user_id`, `date`), - FOREIGN KEY (`user_id`) REFERENCES `user`(`id`) ON UPDATE no action ON DELETE cascade -); ---> statement-breakpoint -INSERT INTO `__new_user_tts_chars`("user_id", "date", "char_count", "created_at", "updated_at") SELECT "user_id", "date", "char_count", "created_at", "updated_at" FROM `user_tts_chars`;--> statement-breakpoint -DROP TABLE `user_tts_chars`;--> statement-breakpoint -ALTER TABLE `__new_user_tts_chars` RENAME TO `user_tts_chars`;--> statement-breakpoint -CREATE INDEX `idx_user_tts_chars_date` ON `user_tts_chars` (`date`); +CREATE INDEX `idx_user_job_events_user_action_created` ON `user_job_events` (`user_id`,`action`,`created_at`); \ No newline at end of file diff --git a/drizzle/sqlite/meta/0010_snapshot.json b/drizzle/sqlite/meta/0010_snapshot.json index 9815696..ed9acdb 100644 --- a/drizzle/sqlite/meta/0010_snapshot.json +++ b/drizzle/sqlite/meta/0010_snapshot.json @@ -1,7 +1,7 @@ { "version": "6", "dialect": "sqlite", - "id": "6f0bbd3d-f8aa-49d1-8565-853f2b2654cb", + "id": "a32ae7d4-c074-44f0-9e90-2473949376ef", "prevId": "2c23f770-269c-4cae-a71e-4ef88355681f", "tables": { "admin_providers": { @@ -1356,21 +1356,7 @@ "isUnique": false } }, - "foreignKeys": { - "user_tts_chars_user_id_user_id_fk": { - "name": "user_tts_chars_user_id_user_id_fk", - "tableFrom": "user_tts_chars", - "tableTo": "user", - "columnsFrom": [ - "user_id" - ], - "columnsTo": [ - "id" - ], - "onDelete": "cascade", - "onUpdate": "no action" - } - }, + "foreignKeys": {}, "compositePrimaryKeys": { "user_tts_chars_user_id_date_pk": { "columns": [ diff --git a/drizzle/sqlite/meta/_journal.json b/drizzle/sqlite/meta/_journal.json index 406dc85..f466fde 100644 --- a/drizzle/sqlite/meta/_journal.json +++ b/drizzle/sqlite/meta/_journal.json @@ -75,7 +75,7 @@ { "idx": 10, "version": "6", - "when": 1780785284041, + "when": 1780794515094, "tag": "0010_user-data-cleanup-cascades", "breakpoints": true } diff --git a/src/app/api/user/claim/route.ts b/src/app/api/user/claim/route.ts index 14cd313..5745cd7 100644 --- a/src/app/api/user/claim/route.ts +++ b/src/app/api/user/claim/route.ts @@ -2,7 +2,7 @@ import { NextRequest, NextResponse } from 'next/server'; import { claimAnonymousData } from '@/lib/server/user/claim-data'; import { auth } from '@/lib/server/auth/auth'; import { db } from '@/db'; -import { audiobooks, documents, userDocumentProgress, userPreferences } from '@/db/schema'; +import { audiobooks, documentSettings, documents, userDocumentProgress, userPreferences } from '@/db/schema'; import { count, eq, ne } from 'drizzle-orm'; import { getOpenReaderTestNamespace, getUnclaimedUserIdForNamespace } from '@/lib/server/testing/test-namespace'; import { errorToLog, serverLogger } from '@/lib/server/logger'; @@ -26,13 +26,14 @@ async function checkClaimMigrationReadiness(): Promise { async function getClaimableCounts( unclaimedUserId: string, -): Promise<{ documents: number; audiobooks: number; preferences: number; progress: number }> { - const [[docCount], [bookCount], [preferencesCount], [progressCount]] = +): Promise<{ documents: number; audiobooks: number; preferences: number; progress: number; documentSettings: number }> { + const [[docCount], [bookCount], [preferencesCount], [progressCount], [settingsCount]] = await Promise.all([ db.select({ count: count() }).from(documents).where(eq(documents.userId, unclaimedUserId)), db.select({ count: count() }).from(audiobooks).where(eq(audiobooks.userId, unclaimedUserId)), db.select({ count: count() }).from(userPreferences).where(eq(userPreferences.userId, unclaimedUserId)), db.select({ count: count() }).from(userDocumentProgress).where(eq(userDocumentProgress.userId, unclaimedUserId)), + db.select({ count: count() }).from(documentSettings).where(eq(documentSettings.userId, unclaimedUserId)), ]); return { @@ -40,6 +41,7 @@ async function getClaimableCounts( audiobooks: Number(bookCount?.count ?? 0), preferences: Number(preferencesCount?.count ?? 0), progress: Number(progressCount?.count ?? 0), + documentSettings: Number(settingsCount?.count ?? 0), }; } diff --git a/src/components/auth/ClaimDataModal.tsx b/src/components/auth/ClaimDataModal.tsx index 54f80ec..7811398 100644 --- a/src/components/auth/ClaimDataModal.tsx +++ b/src/components/auth/ClaimDataModal.tsx @@ -10,6 +10,7 @@ export type ClaimableCounts = { audiobooks: number; preferences: number; progress: number; + documentSettings: number; }; function toClaimableCounts(value: unknown): ClaimableCounts { @@ -19,6 +20,7 @@ function toClaimableCounts(value: unknown): ClaimableCounts { audiobooks: Number(rec.audiobooks ?? 0), preferences: Number(rec.preferences ?? 0), progress: Number(rec.progress ?? 0), + documentSettings: Number(rec.documentSettings ?? 0), }; } @@ -51,7 +53,8 @@ export default function ClaimDataModal({ `Successfully claimed ${claimed.documents} documents, ` + `${claimed.audiobooks} audiobooks, ` + `${claimed.preferences} preference set(s), and ` - + `${claimed.progress} reading progress record(s)!`, + + `${claimed.progress} reading progress record(s) and ` + + `${claimed.documentSettings} document setting(s)!`, ); onClaimed(); router.refresh(); @@ -82,6 +85,7 @@ export default function ClaimDataModal({
  • {claimableCounts.audiobooks} audiobook(s)
  • {claimableCounts.preferences} preference set(s)
  • {claimableCounts.progress} reading progress record(s)
    • +
  • {claimableCounts.documentSettings} document setting(s)
    • diff --git a/src/contexts/OnboardingFlowContext.tsx b/src/contexts/OnboardingFlowContext.tsx index e1cadaf..7d1b5de 100644 --- a/src/contexts/OnboardingFlowContext.tsx +++ b/src/contexts/OnboardingFlowContext.tsx @@ -30,6 +30,7 @@ const EMPTY_CLAIM_COUNTS: ClaimableCounts = { audiobooks: 0, preferences: 0, progress: 0, + documentSettings: 0, }; function toClaimableCounts(value: unknown): ClaimableCounts { @@ -39,6 +40,7 @@ function toClaimableCounts(value: unknown): ClaimableCounts { audiobooks: Number(rec.audiobooks ?? 0), preferences: Number(rec.preferences ?? 0), progress: Number(rec.progress ?? 0), + documentSettings: Number(rec.documentSettings ?? 0), }; } @@ -148,7 +150,11 @@ export function OnboardingFlowProvider({ children }: { children: ReactNode }) { if (isClaimEligible) { claimCounts = await fetchClaimableCounts(); - const total = claimCounts.documents + claimCounts.audiobooks + claimCounts.preferences + claimCounts.progress; + const total = claimCounts.documents + + claimCounts.audiobooks + + claimCounts.preferences + + claimCounts.progress + + claimCounts.documentSettings; claimHasData = total > 0; if (!claimHasData && userId) { claimDismissedUsersRef.current.add(userId); diff --git a/src/db/schema_postgres.ts b/src/db/schema_postgres.ts index e55436d..fe078ee 100644 --- a/src/db/schema_postgres.ts +++ b/src/db/schema_postgres.ts @@ -54,7 +54,8 @@ export const audiobookChapters = pgTable('audiobook_chapters', { // defined here. Only application-specific tables belong in this file. export const userTtsChars = pgTable("user_tts_chars", { - userId: text('user_id').notNull().references(() => user.id, { onDelete: 'cascade' }), + // Also stores device:* and ip:* backstop buckets, so this cannot reference user.id. + userId: text('user_id').notNull(), date: date('date').notNull(), charCount: bigint('char_count', { mode: 'number' }).default(0), createdAt: bigint('created_at', { mode: 'number' }).default(PG_NOW_MS), diff --git a/src/db/schema_sqlite.ts b/src/db/schema_sqlite.ts index 441e884..3f83dfc 100644 --- a/src/db/schema_sqlite.ts +++ b/src/db/schema_sqlite.ts @@ -54,7 +54,8 @@ export const audiobookChapters = sqliteTable('audiobook_chapters', { // defined here. Only application-specific tables belong in this file. export const userTtsChars = sqliteTable("user_tts_chars", { - userId: text('user_id').notNull().references(() => user.id, { onDelete: 'cascade' }), + // Also stores device:* and ip:* backstop buckets, so this cannot reference user.id. + userId: text('user_id').notNull(), date: text('date').notNull(), // SQLite doesn't have native DATE type, text YYYY-MM-DD is standard charCount: integer('char_count').default(0), createdAt: integer('created_at').default(SQLITE_NOW_MS), diff --git a/src/lib/server/auth/auth.ts b/src/lib/server/auth/auth.ts index 53135a3..d0359d1 100644 --- a/src/lib/server/auth/auth.ts +++ b/src/lib/server/auth/auth.ts @@ -164,134 +164,30 @@ const createAuth = () => betterAuth({ import('@/lib/server/user/claim-data'), ]); - // Transfer rate limiting data (TTS char counts) from anonymous user to authenticated user - try { - await rateLimiter.transferAnonymousUsage(anonymousUser.user.id, newUser.user.id); - serverLogger.info({ - event: 'auth.link_account.transfer.rate_limit.succeeded', - anonymousUserIdHash: hashForLog(anonymousUser.user.id), - newUserIdHash: hashForLog(newUser.user.id), - }, 'Transferred rate limit data during account linking'); - } catch (error) { - logDegraded(serverLogger, { - event: 'auth.link_account.transfer.rate_limit.failed', - msg: 'Failed transferring rate limit data during account linking', - step: 'transfer_rate_limit', - context: { - anonymousUserIdHash: hashForLog(anonymousUser.user.id), - newUserIdHash: hashForLog(newUser.user.id), - }, - error, - }); - // Don't throw here to prevent blocking the account linking process - } - - // Transfer audiobooks from anonymous user to new authenticated user - try { - const transferred = await claimData.transferUserAudiobooks(anonymousUser.user.id, newUser.user.id); - if (transferred > 0) { - serverLogger.info({ - event: 'auth.link_account.transfer.audiobooks.succeeded', - transferred, - anonymousUserIdHash: hashForLog(anonymousUser.user.id), - newUserIdHash: hashForLog(newUser.user.id), - }, 'Transferred audiobooks during account linking'); - } - } catch (error) { - logDegraded(serverLogger, { - event: 'auth.link_account.transfer.audiobooks.failed', - msg: 'Failed transferring audiobooks during account linking', - step: 'transfer_audiobooks', - context: { - anonymousUserIdHash: hashForLog(anonymousUser.user.id), - newUserIdHash: hashForLog(newUser.user.id), - }, - error, - }); - // Don't throw here to prevent blocking the account linking process - } - - // Transfer documents from anonymous user to new authenticated user - try { - const transferred = await claimData.transferUserDocuments(anonymousUser.user.id, newUser.user.id); - if (transferred > 0) { - serverLogger.info({ - event: 'auth.link_account.transfer.documents.succeeded', - transferred, - anonymousUserIdHash: hashForLog(anonymousUser.user.id), - newUserIdHash: hashForLog(newUser.user.id), - }, 'Transferred documents during account linking'); - } - } catch (error) { - logDegraded(serverLogger, { - event: 'auth.link_account.transfer.documents.failed', - msg: 'Failed transferring documents during account linking', - step: 'transfer_documents', - context: { - anonymousUserIdHash: hashForLog(anonymousUser.user.id), - newUserIdHash: hashForLog(newUser.user.id), - }, - error, - }); - // Don't throw here to prevent blocking the account linking process - } - - // Transfer preferences from anonymous user to new authenticated user - try { - const transferred = await claimData.transferUserPreferences(anonymousUser.user.id, newUser.user.id); - if (transferred > 0) { - serverLogger.info({ - event: 'auth.link_account.transfer.preferences.succeeded', - transferred, - anonymousUserIdHash: hashForLog(anonymousUser.user.id), - newUserIdHash: hashForLog(newUser.user.id), - }, 'Transferred preferences during account linking'); - } - } catch (error) { - logDegraded(serverLogger, { - event: 'auth.link_account.transfer.preferences.failed', - msg: 'Failed transferring preferences during account linking', - step: 'transfer_preferences', - context: { - anonymousUserIdHash: hashForLog(anonymousUser.user.id), - newUserIdHash: hashForLog(newUser.user.id), - }, - error, - }); - // Don't throw here to prevent blocking the account linking process - } - - // Transfer reading progress from anonymous user to new authenticated user - try { - const transferred = await claimData.transferUserProgress(anonymousUser.user.id, newUser.user.id); - if (transferred > 0) { - serverLogger.info({ - event: 'auth.link_account.transfer.progress.succeeded', - transferred, - anonymousUserIdHash: hashForLog(anonymousUser.user.id), - newUserIdHash: hashForLog(newUser.user.id), - }, 'Transferred reading progress during account linking'); - } - } catch (error) { - logDegraded(serverLogger, { - event: 'auth.link_account.transfer.progress.failed', - msg: 'Failed transferring reading progress during account linking', - step: 'transfer_progress', - context: { - anonymousUserIdHash: hashForLog(anonymousUser.user.id), - newUserIdHash: hashForLog(newUser.user.id), - }, - error, - }); - // Don't throw here to prevent blocking the account linking process - } + const transferred = await claimData.claimAnonymousData( + newUser.user.id, + anonymousUser.user.id, + null, + { cleanupLegacySources: false }, + ); + await rateLimiter.transferAnonymousUsage(anonymousUser.user.id, newUser.user.id); + const { deleteUserStorageData } = await import('@/lib/server/user/data-cleanup'); + await deleteUserStorageData(anonymousUser.user.id, null); + serverLogger.info({ + event: 'auth.link_account.transfer.succeeded', + transferred, + anonymousUserIdHash: hashForLog(anonymousUser.user.id), + newUserIdHash: hashForLog(newUser.user.id), + }, 'Transferred anonymous user data during account linking'); } catch (error) { logServerError(serverLogger, { event: 'auth.link_account.failed', msg: 'onLinkAccount callback failed', error, }); - // Don't throw here to prevent blocking the account linking process + // Better Auth deletes the anonymous user after this callback. + // Block linking when transfer is incomplete so data remains retryable. + throw error; } // Note: Anonymous user will be automatically deleted after this callback completes }, diff --git a/src/lib/server/rate-limit/rate-limiter.ts b/src/lib/server/rate-limit/rate-limiter.ts index b7ce741..1b2d3a0 100644 --- a/src/lib/server/rate-limit/rate-limiter.ts +++ b/src/lib/server/rate-limit/rate-limiter.ts @@ -320,36 +320,34 @@ export class RateLimiter { * Transfer char counts when anonymous user creates an account */ async transferAnonymousUsage(anonymousUserId: string, authenticatedUserId: string): Promise { - const today = new Date().toISOString().split('T')[0]; - const dateValue = today as unknown as UserTtsCharsDateValue; const updatedAt = this.getUpdatedAtValue() as unknown as UserTtsCharsUpdatedAtValue; - const anonymousResult = await safeDb().select({ charCount: userTtsChars.charCount }) - .from(userTtsChars) - .where(and(eq(userTtsChars.userId, anonymousUserId), eq(userTtsChars.date, dateValue))); + await this.runMutation(async (conn) => { + const anonymousRows = await conn.select({ + date: userTtsChars.date, + charCount: userTtsChars.charCount, + }) + .from(userTtsChars) + .where(eq(userTtsChars.userId, anonymousUserId)); - if (anonymousResult.length === 0) return; + for (const anonymousRow of anonymousRows) { + const dateValue = anonymousRow.date as unknown as UserTtsCharsDateValue; + const anonymousCount = Number(anonymousRow.charCount ?? 0); - const anonymousCount = Number(anonymousResult[0].charCount); - - const existingAuth = await safeDb().select({ charCount: userTtsChars.charCount }) - .from(userTtsChars) - .where(and(eq(userTtsChars.userId, authenticatedUserId), eq(userTtsChars.date, dateValue))); - - if (existingAuth.length === 0) { - await safeDb().insert(userTtsChars) - .values({ userId: authenticatedUserId, date: dateValue, charCount: anonymousCount }); - } else { - const existingCount = Number(existingAuth[0].charCount); - if (anonymousCount > existingCount) { - await safeDb().update(userTtsChars) - .set({ charCount: anonymousCount, updatedAt }) - .where(and(eq(userTtsChars.userId, authenticatedUserId), eq(userTtsChars.date, dateValue))); + await conn.insert(userTtsChars) + .values({ userId: authenticatedUserId, date: dateValue, charCount: anonymousCount }) + .onConflictDoUpdate({ + target: [userTtsChars.userId, userTtsChars.date], + set: { + charCount: sql`${userTtsChars.charCount} + ${anonymousCount}`, + updatedAt, + }, + }); } - } - await safeDb().delete(userTtsChars) - .where(and(eq(userTtsChars.userId, anonymousUserId), eq(userTtsChars.date, dateValue))); + await conn.delete(userTtsChars) + .where(eq(userTtsChars.userId, anonymousUserId)); + }); } /** diff --git a/src/lib/server/tts/segments-blobstore.ts b/src/lib/server/tts/segments-blobstore.ts index 94e673b..a6693bd 100644 --- a/src/lib/server/tts/segments-blobstore.ts +++ b/src/lib/server/tts/segments-blobstore.ts @@ -1,4 +1,5 @@ import { + CopyObjectCommand, DeleteObjectsCommand, GetObjectCommand, ListObjectsV2Command, @@ -257,3 +258,38 @@ export async function deleteTtsSegmentPrefix(prefix: string): Promise { return deleted; } + +export async function copyTtsSegmentPrefix(sourcePrefix: string, destinationPrefix: string): Promise { + const cfg = getS3Config(); + const client = getS3ProxyClient(); + const source = sourcePrefix.replace(/^\/+/, ''); + const destination = destinationPrefix.replace(/^\/+/, ''); + if (source === destination) return 0; + + let copied = 0; + let continuationToken: string | undefined; + do { + const listRes = await client.send(new ListObjectsV2Command({ + Bucket: cfg.bucket, + Prefix: source, + ContinuationToken: continuationToken, + })); + const keys = (listRes.Contents ?? []) + .map((item) => item.Key) + .filter((value): value is string => typeof value === 'string' && value.startsWith(source)); + + for (const key of keys) { + await client.send(new CopyObjectCommand({ + Bucket: cfg.bucket, + Key: `${destination}${key.slice(source.length)}`, + CopySource: `${cfg.bucket}/${key}`, + ServerSideEncryption: 'AES256', + })); + copied += 1; + } + + continuationToken = listRes.IsTruncated ? listRes.NextContinuationToken : undefined; + } while (continuationToken); + + return copied; +} diff --git a/src/lib/server/user/claim-data.ts b/src/lib/server/user/claim-data.ts index e36a0db..389429c 100644 --- a/src/lib/server/user/claim-data.ts +++ b/src/lib/server/user/claim-data.ts @@ -1,5 +1,14 @@ import { db } from '@/db'; -import { documents, audiobooks, audiobookChapters, userPreferences, userDocumentProgress } from '@/db/schema'; +import { + documents, + audiobooks, + audiobookChapters, + documentSettings, + ttsSegmentEntries, + ttsSegmentVariants, + userPreferences, + userDocumentProgress, +} from '@/db/schema'; import { eq, and, inArray } from 'drizzle-orm'; import { UNCLAIMED_USER_ID } from '../storage/docstore-legacy'; import { cleanupClaimedLegacyFsSources } from './legacy-fs-claim-cleanup'; @@ -13,6 +22,9 @@ import { isS3Configured } from '../storage/s3'; import { logDegraded } from '../errors/logging'; import { hashForLog, serverLogger } from '../logger'; import { deleteOwnedDocument } from '../documents/delete-owned'; +import { getS3Config } from '../storage/s3'; +import { copyTtsSegmentPrefix } from '../tts/segments-blobstore'; +import { buildTtsSegmentDocumentPrefix } from '../tts/segments'; type AudiobookRow = { id: string; @@ -62,7 +74,7 @@ function contentTypeForAudiobookObject(fileName: string): string { return 'application/octet-stream'; } -async function moveAudiobookBlobScope( +async function copyAudiobookBlobScope( bookId: string, fromUserId: string, toUserId: string, @@ -84,15 +96,27 @@ async function moveAudiobookBlobScope( namespace, ); } +} +async function deleteAudiobookBlobScope( + bookId: string, + userId: string, + namespace: string | null, +): Promise { + const objects = await listAudiobookObjects(bookId, userId, namespace); for (const object of objects) { - await deleteAudiobookObject(bookId, fromUserId, object.fileName, namespace).catch(() => {}); + await deleteAudiobookObject(bookId, userId, object.fileName, namespace); } } -export async function claimAnonymousData(userId: string, unclaimedUserId: string = UNCLAIMED_USER_ID, namespace: string | null = null) { +export async function claimAnonymousData( + userId: string, + unclaimedUserId: string = UNCLAIMED_USER_ID, + namespace: string | null = null, + options?: { cleanupLegacySources?: boolean }, +) { if (!userId) { - return { documents: 0, audiobooks: 0, preferences: 0, progress: 0 }; + return { documents: 0, audiobooks: 0, preferences: 0, progress: 0, documentSettings: 0 }; } const [claimableDocumentRows, claimableAudiobookRows] = await Promise.all([ @@ -106,14 +130,18 @@ export async function claimAnonymousData(userId: string, unclaimedUserId: string .where(eq(audiobooks.userId, unclaimedUserId)) as Promise>, ]); - const [documentsClaimed, audiobooksClaimed, preferencesClaimed, progressClaimed] = await Promise.all([ - transferUserDocuments(unclaimedUserId, userId, { namespace }), + const [documentsClaimed, audiobooksClaimed, preferencesClaimed, progressClaimed, documentSettingsClaimed] = await Promise.all([ + transferUserDocuments(unclaimedUserId, userId, { namespace, transferTts: true }), transferUserAudiobooks(unclaimedUserId, userId, namespace), transferUserPreferences(unclaimedUserId, userId), transferUserProgress(unclaimedUserId, userId), + transferUserDocumentSettings(unclaimedUserId, userId), ]); - if (claimableDocumentRows.length > 0 || claimableAudiobookRows.length > 0) { + if ( + options?.cleanupLegacySources !== false + && (claimableDocumentRows.length > 0 || claimableAudiobookRows.length > 0) + ) { await cleanupClaimedLegacyFsSources({ documentIds: claimableDocumentRows.map((row) => row.id), audiobookIds: claimableAudiobookRows.map((row) => row.id), @@ -140,6 +168,7 @@ export async function claimAnonymousData(userId: string, unclaimedUserId: string audiobooks: audiobooksClaimed, preferences: preferencesClaimed, progress: progressClaimed, + documentSettings: documentSettingsClaimed, }; } @@ -147,7 +176,8 @@ export async function claimAnonymousData(userId: string, unclaimedUserId: string * Transfer documents from one userId to another. * * This is used when an anonymous user upgrades to an authenticated account. - * The underlying blob storage is shared (by sha), so this only moves metadata rows. + * The source document blob is shared, while user-scoped TTS metadata and audio + * are copied before the old ownership is removed. * * @returns number of document rows transferred */ @@ -155,27 +185,46 @@ export async function transferUserDocuments( fromUserId: string, toUserId: string, // eslint-disable-next-line @typescript-eslint/no-explicit-any - options?: { db?: any; namespace?: string | null }, + options?: { db?: any; namespace?: string | null; transferTts?: boolean }, ): Promise { if (!fromUserId || !toUserId) return 0; if (fromUserId === toUserId) return 0; const database = options?.db ?? db; + const storageEnabled = !options?.db && isS3Configured(); const rows = await database.select().from(documents).where(eq(documents.userId, fromUserId)); if (rows.length === 0) return 0; - if (!options?.db && isS3Configured()) { + if (storageEnabled || options?.transferTts) { for (const row of rows) { - await db + await database .insert(documents) .values({ ...row, userId: toUserId }) .onConflictDoNothing(); - await deleteOwnedDocument({ - userId: fromUserId, - documentId: row.id, - namespace: options?.namespace ?? null, - }); + if (options?.transferTts) { + await transferDocumentTtsSegments({ + documentId: row.id, + fromUserId, + toUserId, + namespace: options.namespace ?? null, + database, + copyStorage: storageEnabled, + deleteSourceMetadata: !storageEnabled, + }); + } + if (storageEnabled) { + await deleteOwnedDocument({ + userId: fromUserId, + documentId: row.id, + namespace: options?.namespace ?? null, + }); + } else { + await database.delete(documents).where(and( + eq(documents.userId, fromUserId), + eq(documents.id, row.id), + )); + } } return rows.length; } @@ -190,6 +239,97 @@ export async function transferUserDocuments( return rows.length; } +async function transferDocumentTtsSegments(input: { + documentId: string; + fromUserId: string; + toUserId: string; + namespace: string | null; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + database: any; + copyStorage: boolean; + deleteSourceMetadata: boolean; +}): Promise { + if (input.copyStorage) { + const storagePrefix = getS3Config().prefix; + for (const storageVersion of ['v1', 'v2'] as const) { + await copyTtsSegmentPrefix( + buildTtsSegmentDocumentPrefix({ + storagePrefix, + namespace: input.namespace, + userId: input.fromUserId, + documentId: input.documentId, + storageVersion, + }), + buildTtsSegmentDocumentPrefix({ + storagePrefix, + namespace: input.namespace, + userId: input.toUserId, + documentId: input.documentId, + storageVersion, + }), + ); + } + } + + const entries = await input.database + .select() + .from(ttsSegmentEntries) + .where(and( + eq(ttsSegmentEntries.userId, input.fromUserId), + eq(ttsSegmentEntries.documentId, input.documentId), + )); + const variants = entries.length > 0 + ? await input.database + .select() + .from(ttsSegmentVariants) + .where(and( + eq(ttsSegmentVariants.userId, input.fromUserId), + inArray(ttsSegmentVariants.segmentEntryId, entries.map( + (entry: typeof ttsSegmentEntries.$inferSelect) => entry.segmentEntryId, + )), + )) + : []; + + if (entries.length > 0) { + await input.database.insert(ttsSegmentEntries) + .values(entries.map((entry: typeof ttsSegmentEntries.$inferSelect) => ({ + ...entry, + userId: input.toUserId, + }))) + .onConflictDoNothing(); + } + + const encodedFrom = encodeURIComponent(input.fromUserId); + const encodedTo = encodeURIComponent(input.toUserId); + if (variants.length > 0) { + await input.database.insert(ttsSegmentVariants) + .values(variants.map((variant: typeof ttsSegmentVariants.$inferSelect) => ({ + ...variant, + userId: input.toUserId, + audioKey: variant.audioKey?.replace( + `/users/${encodedFrom}/docs/${input.documentId}/`, + `/users/${encodedTo}/docs/${input.documentId}/`, + ) ?? null, + }))) + .onConflictDoNothing(); + } + + if (input.deleteSourceMetadata) { + if (variants.length > 0) { + await input.database.delete(ttsSegmentVariants).where(and( + eq(ttsSegmentVariants.userId, input.fromUserId), + inArray(ttsSegmentVariants.segmentId, variants.map( + (variant: typeof ttsSegmentVariants.$inferSelect) => variant.segmentId, + )), + )); + } + await input.database.delete(ttsSegmentEntries).where(and( + eq(ttsSegmentEntries.userId, input.fromUserId), + eq(ttsSegmentEntries.documentId, input.documentId), + )); + } +} + /** * Transfer audiobooks from one user to another. * Used when an anonymous user creates a real account. @@ -211,7 +351,7 @@ export async function transferUserAudiobooks( if (isS3Configured()) { for (const book of books) { - await moveAudiobookBlobScope(book.id, fromUserId, toUserId, namespace); + await copyAudiobookBlobScope(book.id, fromUserId, toUserId, namespace); } } @@ -231,6 +371,12 @@ export async function transferUserAudiobooks( .onConflictDoNothing(); } + if (isS3Configured()) { + for (const book of books) { + await deleteAudiobookBlobScope(book.id, fromUserId, namespace); + } + } + await db.delete(audiobookChapters).where(eq(audiobookChapters.userId, fromUserId)); await db.delete(audiobooks).where(eq(audiobooks.userId, fromUserId)); @@ -325,3 +471,44 @@ export async function transferUserProgress(fromUserId: string, toUserId: string) await db.delete(userDocumentProgress).where(eq(userDocumentProgress.userId, fromUserId)); return fromRows.length; } + +export async function transferUserDocumentSettings( + fromUserId: string, + toUserId: string, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + options?: { db?: any }, +): Promise { + if (!fromUserId || !toUserId || fromUserId === toUserId) return 0; + + const database = options?.db ?? db; + const rows = await database + .select() + .from(documentSettings) + .where(eq(documentSettings.userId, fromUserId)); + for (const row of rows) { + const [existing] = await database + .select({ clientUpdatedAtMs: documentSettings.clientUpdatedAtMs }) + .from(documentSettings) + .where(and( + eq(documentSettings.userId, toUserId), + eq(documentSettings.documentId, row.documentId), + )) + .limit(1); + if (existing && Number(existing.clientUpdatedAtMs ?? 0) >= Number(row.clientUpdatedAtMs ?? 0)) { + continue; + } + await database + .insert(documentSettings) + .values({ ...row, userId: toUserId }) + .onConflictDoUpdate({ + target: [documentSettings.documentId, documentSettings.userId], + set: { + dataJson: row.dataJson, + clientUpdatedAtMs: row.clientUpdatedAtMs, + updatedAt: row.updatedAt, + }, + }); + } + await database.delete(documentSettings).where(eq(documentSettings.userId, fromUserId)); + return rows.length; +} diff --git a/tests/unit/transfer-user-documents.vitest.spec.ts b/tests/unit/transfer-user-documents.vitest.spec.ts index f1c0bed..4f512a9 100644 --- a/tests/unit/transfer-user-documents.vitest.spec.ts +++ b/tests/unit/transfer-user-documents.vitest.spec.ts @@ -2,8 +2,13 @@ import { describe, expect, test } from 'vitest'; import Database from 'better-sqlite3'; import { drizzle } from 'drizzle-orm/better-sqlite3'; import { eq } from 'drizzle-orm'; -import { documents } from '../../src/db/schema_sqlite'; -import { transferUserDocuments } from '../../src/lib/server/user/claim-data'; +import { + documentSettings, + documents, + ttsSegmentEntries, + ttsSegmentVariants, +} from '../../src/db/schema_sqlite'; +import { transferUserDocumentSettings, transferUserDocuments } from '../../src/lib/server/user/claim-data'; describe('transferUserDocuments', () => { test('moves document rows to new user without PK conflicts', async () => { @@ -73,4 +78,173 @@ describe('transferUserDocuments', () => { const ids = remainingTo.map((r) => r.id).sort(); expect(ids).toEqual(['doc-a', 'doc-b']); }); + + test('moves document settings while preserving newer destination settings', async () => { + const sqlite = new Database(':memory:'); + sqlite.exec(` + CREATE TABLE document_settings ( + document_id TEXT NOT NULL, + user_id TEXT NOT NULL, + data_json TEXT NOT NULL DEFAULT '{}', + client_updated_at_ms INTEGER NOT NULL DEFAULT 0, + created_at INTEGER, + updated_at INTEGER, + PRIMARY KEY (document_id, user_id) + ); + `); + const settingsDb = drizzle(sqlite); + + await settingsDb.insert(documentSettings).values([ + { + documentId: 'doc-newer-anon', + userId: 'anon', + dataJson: '{"source":"anon"}', + clientUpdatedAtMs: 20, + }, + { + documentId: 'doc-newer-user', + userId: 'anon', + dataJson: '{"source":"anon"}', + clientUpdatedAtMs: 10, + }, + { + documentId: 'doc-newer-user', + userId: 'user', + dataJson: '{"source":"user"}', + clientUpdatedAtMs: 30, + }, + ]); + + const transferred = await transferUserDocumentSettings('anon', 'user', { + db: settingsDb, + }); + expect(transferred).toBe(2); + + const rows = await settingsDb.select().from(documentSettings); + expect(rows).toHaveLength(2); + expect(rows).toEqual(expect.arrayContaining([ + expect.objectContaining({ + documentId: 'doc-newer-anon', + userId: 'user', + dataJson: '{"source":"anon"}', + clientUpdatedAtMs: 20, + }), + expect.objectContaining({ + documentId: 'doc-newer-user', + userId: 'user', + dataJson: '{"source":"user"}', + clientUpdatedAtMs: 30, + }), + ])); + }); + + test('moves TTS metadata without requiring S3', async () => { + const sqlite = new Database(':memory:'); + sqlite.exec(` + CREATE TABLE documents ( + id TEXT NOT NULL, + user_id TEXT NOT NULL, + name TEXT NOT NULL, + type TEXT NOT NULL, + size INTEGER NOT NULL, + last_modified INTEGER NOT NULL, + file_path TEXT NOT NULL, + parse_state TEXT, + parsed_json_key TEXT, + created_at INTEGER, + PRIMARY KEY (id, user_id) + ); + CREATE TABLE tts_segment_entries ( + segment_entry_id TEXT NOT NULL, + user_id TEXT NOT NULL, + document_id TEXT NOT NULL, + reader_type TEXT NOT NULL, + document_version INTEGER NOT NULL, + segment_index INTEGER NOT NULL, + segment_key TEXT, + locator_reader_rank INTEGER NOT NULL, + locator_reader_type TEXT NOT NULL, + locator_page INTEGER NOT NULL, + locator_spine_index INTEGER NOT NULL, + locator_spine_href TEXT NOT NULL, + locator_char_offset INTEGER NOT NULL, + locator_location TEXT NOT NULL, + locator_identity_key TEXT NOT NULL, + text_hash TEXT NOT NULL, + text_length INTEGER NOT NULL DEFAULT 0, + created_at INTEGER, + updated_at INTEGER, + PRIMARY KEY (segment_entry_id, user_id) + ); + CREATE TABLE tts_segment_variants ( + segment_id TEXT NOT NULL, + user_id TEXT NOT NULL, + segment_entry_id TEXT NOT NULL, + settings_hash TEXT NOT NULL, + settings_json TEXT NOT NULL, + audio_key TEXT, + audio_format TEXT NOT NULL DEFAULT 'mp3', + duration_ms INTEGER, + alignment_json TEXT, + status TEXT NOT NULL DEFAULT 'pending', + error TEXT, + created_at INTEGER, + updated_at INTEGER, + PRIMARY KEY (segment_id, user_id) + ); + `); + const transferDb = drizzle(sqlite); + + await transferDb.insert(documents).values({ + id: 'doc', + userId: 'anon', + name: 'doc.pdf', + type: 'pdf', + size: 1, + lastModified: 1, + filePath: 'doc__doc.pdf', + }); + await transferDb.insert(ttsSegmentEntries).values({ + segmentEntryId: 'entry', + userId: 'anon', + documentId: 'doc', + readerType: 'pdf', + documentVersion: 1, + segmentIndex: 0, + locatorReaderRank: 0, + locatorReaderType: 'pdf', + locatorPage: 1, + locatorSpineIndex: 0, + locatorSpineHref: '', + locatorCharOffset: 0, + locatorLocation: '1', + locatorIdentityKey: 'page:1', + textHash: 'hash', + }); + await transferDb.insert(ttsSegmentVariants).values({ + segmentId: 'segment', + userId: 'anon', + segmentEntryId: 'entry', + settingsHash: 'settings', + settingsJson: '{}', + audioKey: 'openreader/tts_segments_v2/users/anon/docs/doc/1/settings/segment.mp3', + status: 'completed', + }); + + await transferUserDocuments('anon', 'user', { + db: transferDb, + transferTts: true, + }); + + expect(await transferDb.select().from(ttsSegmentEntries)).toEqual([ + expect.objectContaining({ segmentEntryId: 'entry', userId: 'user', documentId: 'doc' }), + ]); + expect(await transferDb.select().from(ttsSegmentVariants)).toEqual([ + expect.objectContaining({ + segmentId: 'segment', + userId: 'user', + audioKey: 'openreader/tts_segments_v2/users/user/docs/doc/1/settings/segment.mp3', + }), + ]); + }); }); diff --git a/tests/unit/tts-segments-blobstore.vitest.spec.ts b/tests/unit/tts-segments-blobstore.vitest.spec.ts index e1cb8d9..19be3f2 100644 --- a/tests/unit/tts-segments-blobstore.vitest.spec.ts +++ b/tests/unit/tts-segments-blobstore.vitest.spec.ts @@ -10,7 +10,7 @@ vi.mock('@/lib/server/storage/s3', () => ({ getS3ProxyClient: () => ({ send: mocks.send }), })); -import { deleteTtsSegmentPrefix } from '../../src/lib/server/tts/segments-blobstore'; +import { copyTtsSegmentPrefix, deleteTtsSegmentPrefix } from '../../src/lib/server/tts/segments-blobstore'; describe('TTS segment blob cleanup', () => { beforeEach(() => { @@ -42,4 +42,35 @@ describe('TTS segment blob cleanup', () => { 'Failed deleting 1 TTS segment audio objects', ); }); + + test('copies a user-scoped prefix without deleting its source objects', async () => { + mocks.send + .mockResolvedValueOnce({ + Contents: [{ Key: 'source/doc/audio.mp3' }], + IsTruncated: false, + }) + .mockResolvedValueOnce({}); + + await expect(copyTtsSegmentPrefix('source/doc/', 'destination/doc/')).resolves.toBe(1); + + const copyCommand = mocks.send.mock.calls[1]?.[0]; + expect(copyCommand.constructor.name).toBe('CopyObjectCommand'); + expect(copyCommand.input).toMatchObject({ + Key: 'destination/doc/audio.mp3', + CopySource: 'test-bucket/source/doc/audio.mp3', + }); + expect(mocks.send).toHaveBeenCalledTimes(2); + }); + + test('does not delete source objects when a TTS prefix copy fails', async () => { + mocks.send + .mockResolvedValueOnce({ + Contents: [{ Key: 'source/doc/audio.mp3' }], + IsTruncated: false, + }) + .mockRejectedValueOnce(new Error('copy failed')); + + await expect(copyTtsSegmentPrefix('source/doc/', 'destination/doc/')).rejects.toThrow('copy failed'); + expect(mocks.send).toHaveBeenCalledTimes(2); + }); });