feat(25-01): Zod schemas, upsert service functions, passing tests

- Add upsertGlobalItemSchema and bulkUpsertGlobalItemsSchema to schemas.ts
- Add UpsertGlobalItemInput and BulkUpsertGlobalItemsInput types to types.ts
- Implement upsertGlobalItem with onConflictDoUpdate and tag sync
- Implement bulkUpsertGlobalItems processing array in single transaction
- Fix migration 0003 to only add new columns + unique constraint
- All 21 tests pass including 8 new upsert operation tests

src/server/services/global-item.service.ts

@@ -4,6 +4,7 @@ import { db as prodDb } from "../../db/index.ts";
 import { globalItems, globalItemTags, items, tags } from "../../db/schema.ts";
 
 type Db = typeof prodDb;
+type TxDb = Parameters<Parameters<Db["transaction"]>[0]>[0];
 
 /**
  * Search global items by brand or model and/or tag names.
@@ -71,3 +72,182 @@ export async function getGlobalItemWithOwnerCount(db: Db = prodDb, id: number) {
 
 	return { ...item, ownerCount: result?.ownerCount ?? 0 };
 }
+
+/**
+ * Sync tags for a global item: delete existing, re-insert provided tag names.
+ * Creates tags that don't exist yet (create-if-not-exists).
+ */
+async function syncGlobalItemTags(
+	tx: TxDb,
+	globalItemId: number,
+	tagNames: string[],
+) {
+	await tx
+		.delete(globalItemTags)
+		.where(eq(globalItemTags.globalItemId, globalItemId));
+
+	for (const name of tagNames) {
+		const [tag] = await tx
+			.insert(tags)
+			.values({ name })
+			.onConflictDoUpdate({ target: tags.name, set: { name } })
+			.returning({ id: tags.id });
+
+		await tx.insert(globalItemTags).values({ globalItemId, tagId: tag.id });
+	}
+}
+
+/**
+ * Upsert a single global item by (brand, model).
+ * Creates if not exists, updates all non-key fields if exists.
+ * Tag sync: provided → sync; undefined → leave untouched; [] → clear all tags.
+ */
+export async function upsertGlobalItem(
+	db: Db,
+	data: {
+		brand: string;
+		model: string;
+		category?: string;
+		weightGrams?: number;
+		priceCents?: number;
+		imageUrl?: string;
+		description?: string;
+		sourceUrl?: string;
+		imageCredit?: string;
+		imageSourceUrl?: string;
+		tags?: string[];
+	},
+) {
+	return await db.transaction(async (tx) => {
+		const [existing] = await tx
+			.select({ id: globalItems.id })
+			.from(globalItems)
+			.where(
+				and(
+					eq(globalItems.brand, data.brand),
+					eq(globalItems.model, data.model),
+				),
+			);
+
+		const { tags: tagNames, ...itemData } = data;
+
+		const [item] = await tx
+			.insert(globalItems)
+			.values({
+				brand: itemData.brand,
+				model: itemData.model,
+				category: itemData.category ?? null,
+				weightGrams: itemData.weightGrams ?? null,
+				priceCents: itemData.priceCents ?? null,
+				imageUrl: itemData.imageUrl ?? null,
+				description: itemData.description ?? null,
+				sourceUrl: itemData.sourceUrl ?? null,
+				imageCredit: itemData.imageCredit ?? null,
+				imageSourceUrl: itemData.imageSourceUrl ?? null,
+			})
+			.onConflictDoUpdate({
+				target: [globalItems.brand, globalItems.model],
+				set: {
+					category: itemData.category ?? null,
+					weightGrams: itemData.weightGrams ?? null,
+					priceCents: itemData.priceCents ?? null,
+					imageUrl: itemData.imageUrl ?? null,
+					description: itemData.description ?? null,
+					sourceUrl: itemData.sourceUrl ?? null,
+					imageCredit: itemData.imageCredit ?? null,
+					imageSourceUrl: itemData.imageSourceUrl ?? null,
+				},
+			})
+			.returning();
+
+		if (tagNames !== undefined) {
+			await syncGlobalItemTags(tx, item.id, tagNames);
+		}
+
+		return { item, created: !existing };
+	});
+}
+
+/**
+ * Bulk upsert global items in a single transaction.
+ * Returns { created, updated, items } with accurate counts.
+ * Rolls back entirely if any item fails.
+ */
+export async function bulkUpsertGlobalItems(
+	db: Db,
+	itemsData: Array<{
+		brand: string;
+		model: string;
+		category?: string;
+		weightGrams?: number;
+		priceCents?: number;
+		imageUrl?: string;
+		description?: string;
+		sourceUrl?: string;
+		imageCredit?: string;
+		imageSourceUrl?: string;
+		tags?: string[];
+	}>,
+) {
+	return await db.transaction(async (tx) => {
+		let created = 0;
+		let updated = 0;
+		const resultItems = [];
+
+		for (const data of itemsData) {
+			const [existing] = await tx
+				.select({ id: globalItems.id })
+				.from(globalItems)
+				.where(
+					and(
+						eq(globalItems.brand, data.brand),
+						eq(globalItems.model, data.model),
+					),
+				);
+
+			const { tags: tagNames, ...itemData } = data;
+
+			const [item] = await tx
+				.insert(globalItems)
+				.values({
+					brand: itemData.brand,
+					model: itemData.model,
+					category: itemData.category ?? null,
+					weightGrams: itemData.weightGrams ?? null,
+					priceCents: itemData.priceCents ?? null,
+					imageUrl: itemData.imageUrl ?? null,
+					description: itemData.description ?? null,
+					sourceUrl: itemData.sourceUrl ?? null,
+					imageCredit: itemData.imageCredit ?? null,
+					imageSourceUrl: itemData.imageSourceUrl ?? null,
+				})
+				.onConflictDoUpdate({
+					target: [globalItems.brand, globalItems.model],
+					set: {
+						category: itemData.category ?? null,
+						weightGrams: itemData.weightGrams ?? null,
+						priceCents: itemData.priceCents ?? null,
+						imageUrl: itemData.imageUrl ?? null,
+						description: itemData.description ?? null,
+						sourceUrl: itemData.sourceUrl ?? null,
+						imageCredit: itemData.imageCredit ?? null,
+						imageSourceUrl: itemData.imageSourceUrl ?? null,
+					},
+				})
+				.returning();
+
+			if (tagNames !== undefined) {
+				await syncGlobalItemTags(tx, item.id, tagNames);
+			}
+
+			if (existing) {
+				updated++;
+			} else {
+				created++;
+			}
+			resultItems.push(item);
+		}
+
+		return { created, updated, items: resultItems };
+	});
+}