clippr / backend / src / network / commit.ts

/*
 * clippr: a social bookmarking service for the AT Protocol
 * Copyright (c) 2025 clippr contributors.
 * SPDX-License-Identifier: AGPL-3.0-only
 */

import type { CommitEvent } from "@skyware/jetstream";
import { Database } from "../db/database.js";
import { clipsTable, tagsTable, usersTable } from "../db/schema.js";
import { is } from "@atcute/lexicons";
import {
	SocialClipprActorProfile,
	SocialClipprFeedClip,
	SocialClipprFeedTag,
} from "@clipprjs/lexicons";
import Logger from "../logger.js";
import { isBlob } from "@atcute/lexicons/interfaces";
import { validateClip, validateProfile, validateTag } from "./validator.js";
import { convertDidToString } from "./converters.js";
import { hashString } from "../hasher.js";
import { and, eq } from "drizzle-orm";
import type { TagRef } from "../api/types.js";

const db = Database.getInstance().getDb();

/// Converts a microsecond Unix date to a Date object, for type reasons.
function convertMicroToDate(micro: number): Date {
	return new Date(micro / 1000);
}

export async function handleClip(
	event: CommitEvent<`social.clippr.${string}`>,
): Promise<void> {
	if (event.commit.operation === "delete") {
		await db
			.delete(clipsTable)
			.where(
				and(
					eq(clipsTable.did, event.did),
					eq(clipsTable.recordKey, event.commit.rkey),
				),
			);
		Logger.verbose(`Deleted clip: ${event.did}/${event.commit.rkey}`, event);
		return;
	}

	if (event.commit.record.$type !== "social.clippr.feed.clip") {
		Logger.verbose(
			`Mismatched type for incoming clip record (${event.did}/${event.commit.rkey})`,
			event.commit.record,
		);
	}

	if (!is(SocialClipprFeedClip.mainSchema, event.commit.record)) {
		Logger.verbose(
			`Invalid schema for incoming clip record (${event.did}/${event.commit.rkey})`,
			event.commit.record,
		);
		return;
	}

	const record: SocialClipprFeedClip.Main = {
		$type: "social.clippr.feed.clip",
		createdAt: event.commit.record.createdAt,
		description: event.commit.record.description,
		languages: event.commit.record.languages,
		notes: event.commit.record.notes,
		tags: event.commit.record.tags,
		title: event.commit.record.title,
		unlisted: event.commit.record.unlisted,
		unread: event.commit.record.unread,
		url: event.commit.record.url,
	};

	// xxh64, NOT xxh3 learned that the hard way
	const urlHash: string = await hashString(record.url);

	if (urlHash !== event.commit.rkey) {
		Logger.verbose(
			`Record key hash (${event.commit.rkey}) does not match hash of URL (${urlHash}) in incoming clip record (${event.did})`,
			event.commit.record,
		);
		return;
	}

	if (!(await validateClip(record))) return;

	if (event.commit.operation === "update") {
		await db
			.update(clipsTable)
			.set({
				did: convertDidToString(event.did),
				cid: event.commit.cid,
				timestamp: convertMicroToDate(event.time_us),
				recordKey: event.commit.rkey,
				createdAt: new Date(record.createdAt),
				indexedAt: new Date(),
				url: record.url,
				title: record.title,
				description: record.description,
				tags: record.tags as TagRef[] | undefined,
				notes: record.notes,
				unlisted: record.unlisted,
				unread: record.unread,
				languages: record.languages,
			})
			.where(
				and(
					eq(clipsTable.did, event.did),
					eq(clipsTable.recordKey, event.commit.rkey),
				),
			);
		Logger.verbose(`Updated clip: ${event.did}/${event.commit.rkey}`, event);
		return;
	}

	await db.insert(clipsTable).values({
		// @ts-expect-error Weird type error despite being a normal string.
		did: convertDidToString(event.did),
		cid: event.commit.cid,
		timestamp: convertMicroToDate(event.time_us),
		recordKey: event.commit.rkey,
		createdAt: new Date(record.createdAt),
		indexedAt: new Date(),
		url: record.url,
		title: record.title,
		description: record.description,
		tags: record.tags,
		notes: record.notes,
		unlisted: record.unlisted,
		unread: record.unread,
		languages: record.languages,
	});

	Logger.verbose(`Indexed new clip: ${event.did}/${event.commit.rkey}`, event);
}

export async function handleTag(
	event: CommitEvent<`social.clippr.${string}`>,
): Promise<void> {
	if (event.commit.operation === "delete") {
		await db
			.delete(tagsTable)
			.where(
				and(
					eq(tagsTable.did, event.did),
					eq(tagsTable.recordKey, event.commit.rkey),
				),
			);
		Logger.verbose(`Deleted tag: ${event.did}/${event.commit.rkey}`, event);
		return;
	}

	if (event.commit.record.$type !== "social.clippr.feed.tag") {
		Logger.verbose(
			`Mismatched type for incoming tag record (${event.did}/${event.commit.rkey})`,
			event.commit.record,
		);
		return;
	}

	if (!is(SocialClipprFeedTag.mainSchema, event.commit.record)) {
		Logger.verbose(
			`Invalid schema for incoming tag record (${event.did}/${event.commit.rkey})`,
			event.commit.record,
		);
		return;
	}

	const record: SocialClipprFeedTag.Main = {
		$type: "social.clippr.feed.tag",
		createdAt: event.commit.record.createdAt,
		name: event.commit.record.name,
		color: event.commit.record.color,
		description: event.commit.record.description,
	};

	if (record.name !== event.commit.rkey) {
		Logger.verbose(
			`Record key does not match name of incoming tag record (${event.did}/${event.commit.rkey})`,
			event.commit.record,
		);
		return;
	}

	// Independent validations
	if (!(await validateTag(record))) {
		return;
	}

	if (event.commit.operation === "update") {
		await db
			.update(tagsTable)
			.set({
				timestamp: convertMicroToDate(event.time_us),
				did: convertDidToString(event.did),
				cid: event.commit.cid,
				recordKey: event.commit.rkey,
				name: record.name,
				description: record.description,
				color: record.color,
				createdAt: new Date(record.createdAt),
				indexedAt: new Date(),
			})
			.where(
				and(
					eq(tagsTable.did, event.did),
					eq(tagsTable.recordKey, event.commit.rkey),
				),
			);
		Logger.verbose(`Updated tag: ${event.did}/${event.commit.rkey}`, event);
		return;
	}

	await db.insert(tagsTable).values({
		timestamp: convertMicroToDate(event.time_us),
		did: convertDidToString(event.did),
		cid: event.commit.cid,
		recordKey: event.commit.rkey,
		name: record.name,
		description: record.description,
		color: record.color,
		createdAt: new Date(record.createdAt),
		indexedAt: new Date(),
	});

	Logger.verbose(`Indexed new tag: ${event.did}/${event.commit.rkey}`, event);
}

export async function handleProfile(
	event: CommitEvent<`social.clippr.${string}`>,
): Promise<void> {
	if (event.commit.operation === "delete") {
		await db.delete(usersTable).where(eq(usersTable.did, event.did));
		Logger.verbose(`Deleted profile: ${event.did}`, event);
		return;
	}

	if (event.commit.record.$type !== "social.clippr.actor.profile") {
		Logger.verbose(
			`Mismatched type for incoming profile record (${event.did})`,
			event.commit.record,
		);
		return;
	}

	if (!is(SocialClipprActorProfile.mainSchema, event.commit.record)) {
		Logger.verbose(
			`Invalid schema for incoming profile record (${event.did})`,
			event.commit.record,
		);
		return;
	}

	const record: SocialClipprActorProfile.Main = {
		$type: "social.clippr.actor.profile",
		createdAt: event.commit.record.createdAt,
		displayName: event.commit.record.displayName,
		description: event.commit.record.description || undefined,
		avatar: event.commit.record.avatar || undefined,
	};

	if (event.commit.rkey !== "self") {
		Logger.verbose(
			`Record key of incoming profile record does not match 'self' (${event.did})`,
			event.commit.record,
		);
		return;
	}

	// This needs to be here so the avatar can be recognized as a proper blob.
	if (record.avatar) {
		if (!isBlob(record.avatar)) {
			Logger.verbose(
				`Avatar in incoming profile record is not a blob (${event.did})`,
				record,
			);
			return;
		}

		if (record.avatar.mimeType.match(/^image\/(png|jpeg)$/i) === null) {
			Logger.verbose(
				`Avatar in incoming profile record is not a PNG or JPEG (${event.did})`,
				record,
			);
			return;
		}

		if (record.avatar.ref?.$link === undefined) {
			Logger.verbose(
				`Avatar in incoming profile record has no link to blob (${event.did})`,
				record,
			);
			return;
		}

		if (record.avatar.size > 1000000) {
			Logger.verbose(
				`Avatar in incoming profile record is too large (${event.did})`,
				record,
			);
			return;
		}
	}

	// Independent validations
	if (!(await validateProfile(record))) {
		return;
	}

	if (event.commit.operation === "update") {
		await db
			.update(usersTable)
			.set({
				did: convertDidToString(event.did),
				cid: event.commit.cid,
				timestamp: convertMicroToDate(event.time_us),
				createdAt: new Date(record.createdAt),
				displayName: record.displayName,
				avatar: record.avatar?.ref.$link,
				description: record.description,
			})
			.where(eq(usersTable.did, convertDidToString(event.did)));
		Logger.verbose(`Updated profile: ${convertDidToString(event.did)}`, event);
		return;
	}

	await db.insert(usersTable).values({
		did: convertDidToString(event.did),
		cid: event.commit.cid,
		timestamp: convertMicroToDate(event.time_us),
		createdAt: new Date(record.createdAt),
		displayName: record.displayName,
		avatar: record.avatar?.ref.$link,
		description: record.description,
	});

	Logger.verbose(
		`Indexed new profile: ${convertDidToString(event.did)}`,
		event,
	);
}