perf: batch Loops profile checks and index cap delete, recovery and caps list queries

.github/workflows/database-profile-reliability.yml

@@ -53,6 +53,7 @@ jobs:
         run: >-
           bun test
           scripts/loops/activation.test.ts
+          scripts/loops/batch.test.ts
           scripts/loops/sources.test.ts
           scripts/loops/profile.test.ts
           scripts/loops/lifecycle.test.ts

apps/web/app/(org)/dashboard/caps/page.tsx

@@ -151,6 +151,23 @@ export default async function CapsPage(props: PageProps<"/dashboard/caps">) {
 			),
 		);
 
+	const capsPage = db()
+		.select({ id: videos.id })
+		.from(videos)
+		.leftJoin(organizations, eq(videos.orgId, organizations.id))
+		.where(
+			and(
+				eq(videos.ownerId, userId),
+				eq(videos.orgId, user.activeOrganizationId),
+				isNull(videos.folderId),
+				isNull(organizations.tombstoneAt),
+			),
+		)
+		.orderBy(desc(videos.effectiveCreatedAt), desc(videos.id))
+		.limit(limit)
+		.offset(offset)
+		.as("caps_page");
+
 	const videoDataPromise = db()
 		.select({
 			id: videos.id,
@@ -192,18 +209,11 @@ export default async function CapsPage(props: PageProps<"/dashboard/caps">) {
 			settings: videos.settings,
 		})
 		.from(videos)
+		.innerJoin(capsPage, eq(videos.id, capsPage.id))
 		.leftJoin(comments, eq(videos.id, comments.videoId))
 		.leftJoin(organizations, eq(videos.orgId, organizations.id))
 		.leftJoin(users, eq(videos.ownerId, users.id))
 		.leftJoin(videoUploads, eq(videos.id, videoUploads.videoId))
-		.where(
-			and(
-				eq(videos.ownerId, userId),
-				eq(videos.orgId, user.activeOrganizationId),
-				isNull(videos.folderId),
-				isNull(organizations.tombstoneAt),
-			),
-		)
 		.groupBy(
 			videos.id,
 			videos.ownerId,
@@ -219,9 +229,7 @@ export default async function CapsPage(props: PageProps<"/dashboard/caps">) {
 			videos.orgId,
 			users.name,
 		)
-		.orderBy(desc(videos.effectiveCreatedAt))
-		.limit(limit)
-		.offset(offset);
+		.orderBy(desc(videos.effectiveCreatedAt), desc(videos.id));
 
 	const foldersDataPromise = db()
 		.select({

packages/database/loops/activation.ts

@@ -1,11 +1,21 @@
+import { placeholders } from "./sql";
+
 export const legacyActivationQuery = `
 	SELECT
 		EXISTS(SELECT 1 FROM videos WHERE ownerId = ? LIMIT 1) AS hasVideo,
 		EXISTS(SELECT 1 FROM videos WHERE ownerId = ? AND public = 1 LIMIT 1) AS hasSharedVideo
 `;
 
-export const activationQuery = `
+export const batchLegacyActivationQuery = (users: number) => `
 	SELECT
+		u.id AS userId,
+		EXISTS(SELECT 1 FROM videos WHERE ownerId = u.id LIMIT 1) AS hasVideo,
+		EXISTS(SELECT 1 FROM videos WHERE ownerId = u.id AND public = 1 LIMIT 1) AS hasSharedVideo
+	FROM users u
+	WHERE u.id IN (${placeholders(users)})
+`;
+
+const activationSignals = `
 		MAX(CASE WHEN u.video_id IS NOT NULL OR (j.video_id IS NOT NULL AND j.state <> 'verified') THEN 1 ELSE 0 END) AS hasPendingUpload,
 		MAX(CASE WHEN u.video_id IS NULL
 			AND v.duration > 0
@@ -24,6 +34,15 @@ export const activationQuery = `
 		MAX(GREATEST(DATE_ADD(v.createdAt, INTERVAL 5 MINUTE), COALESCE(v.firstViewEmailSentAt, v.createdAt))) AS lastActivationNotificationAt
 	FROM videos v
 	LEFT JOIN video_uploads u ON u.video_id = v.id
-	LEFT JOIN video_processing_jobs j ON j.video_id = v.id
+	LEFT JOIN video_processing_jobs j ON j.video_id = v.id`;
+
+export const activationQuery = `
+	SELECT ${activationSignals}
 	WHERE v.ownerId = ? AND v.isScreenshot = 0
 `;
+
+export const batchActivationQuery = (users: number) => `
+	SELECT v.ownerId AS userId, ${activationSignals}
+	WHERE v.ownerId IN (${placeholders(users)}) AND v.isScreenshot = 0
+	GROUP BY v.ownerId
+`;

packages/database/loops/jobs.ts

@@ -0,0 +1,121 @@
+import type {
+	Connection,
+	ResultSetHeader,
+	RowDataPacket,
+} from "mysql2/promise";
+import { placeholders } from "./sql";
+
+export type Job = RowDataPacket & {
+	userId: string;
+	revision: number;
+	failures: number;
+	syncedEmail: string | null;
+	lastSyncedAt: string | null;
+	profileHash: string | null;
+	teammateJoinedAt: string | null;
+};
+
+export async function claimJobs(
+	database: Connection,
+	token: string,
+	allowedUserIds: string[] | null,
+	limit: number,
+) {
+	await database.beginTransaction();
+	try {
+		const [jobs] = await database.query<Job[]>(
+			`SELECT * FROM loops_sync_jobs WHERE nextAttemptAt<=UTC_TIMESTAMP() AND (leaseUntil IS NULL OR leaseUntil<UTC_TIMESTAMP()) ${allowedUserIds ? "AND userId IN (?)" : ""} ORDER BY nextAttemptAt LIMIT ? FOR UPDATE SKIP LOCKED`,
+			allowedUserIds ? [allowedUserIds, limit] : [limit],
+		);
+		if (jobs.length)
+			await database.execute(
+				`UPDATE loops_sync_jobs SET leaseToken=?,leaseUntil=DATE_ADD(UTC_TIMESTAMP(),INTERVAL 2 MINUTE) WHERE userId IN (${placeholders(jobs.length)})`,
+				[token, ...jobs.map((job) => job.userId)],
+			);
+		await database.commit();
+		return jobs;
+	} catch (error) {
+		await database.rollback();
+		throw error;
+	}
+}
+
+export async function renewLease(
+	database: Connection,
+	job: Pick<Job, "userId">,
+	token: string,
+) {
+	const [result] = await database.execute<ResultSetHeader>(
+		"UPDATE loops_sync_jobs SET leaseUntil=DATE_ADD(UTC_TIMESTAMP(),INTERVAL 2 MINUTE) WHERE userId=? AND leaseToken=?",
+		[job.userId, token],
+	);
+	return result.affectedRows > 0;
+}
+
+export async function releaseJobs(
+	database: Connection,
+	token: string,
+	userIds: string[],
+) {
+	await database.execute(
+		`UPDATE loops_sync_jobs SET leaseToken=NULL,leaseUntil=NULL WHERE leaseToken=? AND userId IN (${placeholders(userIds.length)})`,
+		[token, ...userIds],
+	);
+}
+
+export async function deferJob(
+	database: Connection,
+	job: Pick<Job, "userId" | "revision">,
+	token: string,
+	reason: string,
+	nextAttemptAt: Date,
+) {
+	await database.execute(
+		"UPDATE loops_sync_jobs SET failures=CASE WHEN revision=? THEN failures+1 ELSE 0 END,lastError=CASE WHEN revision=? THEN ? ELSE NULL END,nextAttemptAt=CASE WHEN revision=? THEN ? ELSE UTC_TIMESTAMP() END,leaseToken=NULL,leaseUntil=NULL WHERE userId=? AND leaseToken=?",
+		[
+			job.revision,
+			job.revision,
+			reason,
+			job.revision,
+			nextAttemptAt,
+			job.userId,
+			token,
+		],
+	);
+}
+
+export async function completeUnchangedJob(
+	database: Connection,
+	job: Pick<Job, "userId" | "revision">,
+	token: string,
+	nextAttemptAt: Date,
+) {
+	await database.execute(
+		"UPDATE loops_sync_jobs SET failures=0,lastError=NULL,nextAttemptAt=CASE WHEN revision=? THEN ? ELSE UTC_TIMESTAMP() END,leaseToken=NULL,leaseUntil=NULL WHERE userId=? AND leaseToken=?",
+		[job.revision, nextAttemptAt, job.userId, token],
+	);
+}
+
+export async function completeUnchangedJobs(
+	database: Connection,
+	token: string,
+	jobs: { job: Pick<Job, "userId" | "revision">; nextAttemptAt: Date }[],
+) {
+	const groups = new Map<number, typeof jobs>();
+	for (const entry of jobs) {
+		const key = entry.nextAttemptAt.getTime();
+		const group = groups.get(key);
+		if (group) group.push(entry);
+		else groups.set(key, [entry]);
+	}
+	for (const group of groups.values())
+		await database.execute(
+			`UPDATE loops_sync_jobs SET failures=0,lastError=NULL,nextAttemptAt=CASE WHEN (userId,revision) IN (${group.map(() => "(?,?)").join(",")}) THEN ? ELSE UTC_TIMESTAMP() END,leaseToken=NULL,leaseUntil=NULL WHERE leaseToken=? AND userId IN (${placeholders(group.length)})`,
+			[
+				...group.flatMap(({ job }) => [job.userId, job.revision]),
+				group[0].nextAttemptAt,
+				token,
+				...group.map(({ job }) => job.userId),
+			],
+		);
+}