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); + }); });