Merge branch 'main' into kalilsn/form-switcher

Author: kalilsn

core/actions/_lib/runActionInstance.ts

@@ -52,6 +52,7 @@ const getActionInstance = (actionInstanceId: ActionInstancesId, pubId: PubsId, t
 			"updatedAt",
 			"stageId",
 			"action",
+			"name",
 			// this is to check whether the pub is still in the stage the actionInstance is in
 			// often happens when an action is scheduled but a pub is moved before the action runs
 			jsonObjectFrom(
@@ -181,6 +182,7 @@ const _runActionInstance = async (
 			lastModifiedBy,
 			actionRunId: args.actionRunId,
 			userId: isActionUserInitiated ? args.userId : undefined,
+			actionInstance: args.actionInstance,
 		});
 
 		if (isClientExceptionOptions(result)) {

core/actions/corePubFields.ts

@@ -3,6 +3,7 @@ import { afterEach } from "node:test";
 import { describe, expect, it, vitest } from "vitest";
 
 import type {
+	ActionInstancesId,
 	ActionRunsId,
 	CommunitiesId,
 	PubFieldsId,
@@ -11,7 +12,7 @@ import type {
 	PubValuesId,
 	StagesId,
 } from "db/public";
-import { CoreSchemaType } from "db/public";
+import { Action, CoreSchemaType } from "db/public";
 
 import type { ActionPub, RunProps } from "../types";
 import type { action } from "./action";
@@ -133,6 +134,15 @@ const pub = {
 } as ActionPub;
 
 const RUN_OPTIONS: RunProps<typeof action> = {
+	actionInstance: {
+		id: "" as ActionInstancesId,
+		name: "deposit to datacite",
+		stageId: "" as StagesId,
+		createdAt: new Date(),
+		updatedAt: new Date(),
+		action: Action.datacite,
+		config: {},
+	},
 	actionRunId: "" as ActionRunsId,
 	stageId: "" as StagesId,
 	communityId: "" as CommunitiesId,

core/actions/datacite/run.test.ts

@@ -5,7 +5,7 @@ import { logger } from "logger";
 import type { action } from "./action";
 import { defineRun } from "../types";
 
-export const run = defineRun<typeof action>(async ({ pub, config, args }) => {
+export const run = defineRun<typeof action>(async ({ actionInstance, pub, config, args }) => {
 	logger.info({
 		msg: `Logging${args?.text ? ` ${args.text}` : ""}`,
 		pub,
@@ -16,6 +16,7 @@ export const run = defineRun<typeof action>(async ({ pub, config, args }) => {
 	return {
 		success: true,
 		report: `Logged out ${args?.text || "some data"}, check your console.`,
+		title: `Successfully ran ${actionInstance.name}`,
 		data: {},
 	};
 });

core/actions/log/run.ts

@@ -13,6 +13,10 @@ export const action = defineAction({
 			authToken: z.string().describe("PubPub v6 API auth token"),
 			title: z.string().describe("Title of the Pub"),
 			content: z.string().describe("Content of the Pub"),
+			idField: z
+				.string()
+				.regex(/\w+:\w+/)
+				.describe("Field on this pub to write to id to|ID Field"),
 		}),
 	},
 	description: "Sync a PubPub Platform pub to v6",

core/actions/pushToV6/action.ts

@@ -11,7 +11,6 @@ import type { ClientExceptionOptions } from "~/lib/serverActions";
 import { db } from "~/kysely/database";
 import { autoRevalidate } from "~/lib/server/cache/autoRevalidate";
 import { isClientExceptionOptions } from "~/lib/serverActions";
-import * as corePubFields from "../corePubFields";
 import { defineRun } from "../types";
 
 const authError = {
@@ -159,14 +158,16 @@ const updateV6PubText = async (
 	}
 };
 
-const updateV6PubId = async (pubId: PubsId, v6PubId: string, lastModifiedBy: LastModifiedBy) => {
+const updateV6PubId = async (
+	pubId: PubsId,
+	v6PubId: string,
+	lastModifiedBy: LastModifiedBy,
+	idField: string
+) => {
 	await autoRevalidate(
 		db
 			.with("field", (db) =>
-				db
-					.selectFrom("pub_fields")
-					.select("id")
-					.where("slug", "=", corePubFields.v6PubId.slug)
+				db.selectFrom("pub_fields").select("id").where("slug", "=", idField)
 			)
 			.insertInto("pub_values")
 			.values((eb) => ({
@@ -188,7 +189,7 @@ export const run = defineRun<typeof action>(async ({ pub, config, args, lastModi
 
 		let v6Pub: { id: string };
 
-		const v6PubId = pub.values.find((value) => value.fieldSlug === corePubFields.v6PubId.slug)
+		const v6PubId = pub.values.find((value) => value.fieldSlug === config.idField)
 			?.value as string;
 		// Fetch the pub if the v7 pub already had a v6 pub id
 		if (v6PubId) {
@@ -220,7 +221,7 @@ export const run = defineRun<typeof action>(async ({ pub, config, args, lastModi
 			v6Pub = createV6PubResponse;
 
 			// Update the v6 pub id in the v7 pub
-			await updateV6PubId(pub.id, v6Pub.id, lastModifiedBy);
+			await updateV6PubId(pub.id, v6Pub.id, lastModifiedBy, config.idField);
 		}
 
 		// Update the v6 pub content

core/actions/pushToV6/run.ts

@@ -14,11 +14,8 @@ import type { Dependency, FieldConfig, FieldConfigItem } from "ui/auto-form";
 import type * as Icons from "ui/icon";
 import { Event } from "db/public";
 
-import type { CorePubField } from "./corePubFields";
 import type { ClientExceptionOptions } from "~/lib/serverActions";
 
-export type ActionPubType = CorePubField[];
-
 type ZodObjectOrWrapped = z.ZodObject<any, any> | z.ZodEffects<z.ZodObject<any, any>>;
 export type ZodObjectOrWrappedOrOptional = ZodObjectOrWrapped | z.ZodOptional<ZodObjectOrWrapped>;
 
@@ -28,11 +25,7 @@ export type ActionPub = ProcessedPub<{
 }>;
 
 export type RunProps<T extends Action> =
-	T extends Action<
-		infer P extends ActionPubType,
-		infer C,
-		infer A extends ZodObjectOrWrappedOrOptional
-	>
+	T extends Action<infer C, infer A extends ZodObjectOrWrappedOrOptional>
 		? {
 				config: C["_output"] & { pubFields: { [K in keyof C["_output"]]?: string[] } };
 				configFieldOverrides: Set<string>;
@@ -52,6 +45,7 @@ export type RunProps<T extends Action> =
 				 * The user ID of the user who initiated the action, if any
 				 */
 				userId?: UsersId;
+				actionInstance: ActionInstances;
 			}
 		: never;
 
@@ -66,7 +60,6 @@ export type TokenDef = {
 };
 
 export type Action<
-	P extends ActionPubType = ActionPubType,
 	C extends ZodObjectOrWrapped = ZodObjectOrWrapped,
 	A extends ZodObjectOrWrappedOrOptional = ZodObjectOrWrappedOrOptional,
 	N extends ActionName = ActionName,
@@ -138,12 +131,11 @@ export type Action<
 };
 
 export const defineAction = <
-	T extends ActionPubType,
 	C extends ZodObjectOrWrapped,
 	A extends ZodObjectOrWrappedOrOptional,
 	N extends ActionName,
 >(
-	action: Action<T, C, A, N>
+	action: Action<C, A, N>
 ) => action;
 
 export type ActionSuccess = {

core/actions/types.ts

@@ -0,0 +1,102 @@
+import type { NextRequest } from "next/server";
+import type { PoolClient } from "pg";
+
+import { createSSEHandler } from "use-next-sse";
+
+import type { PublicSchema } from "db/public";
+import type { databaseTableNames } from "db/table-names";
+import { logger } from "logger";
+
+import { pool } from "~/kysely/database";
+import { getLoginData } from "~/lib/authentication/loginData";
+import { findCommunityBySlug } from "~/lib/server/community";
+
+type Tables = (typeof databaseTableNames)[number];
+
+/**
+ * Tables that are currently supported for SSE notifications
+ */
+const notifyTables = ["action_runs"] as const satisfies Tables[];
+export type NotifyTables = (typeof notifyTables)[number];
+
+const parseNotifyTables = (tables: string[]): NotifyTables[] => {
+	return tables.filter((table): table is NotifyTables =>
+		notifyTables.includes(table as NotifyTables)
+	);
+};
+
+export const dynamic = "force-dynamic";
+export type ChangeNotification<T extends NotifyTables> = {
+	table: T;
+	operation: "insert" | "update" | "delete";
+	row: PublicSchema[T];
+};
+
+const handleClose = (client?: PoolClient) => {
+	return () => {
+		logger.info("closing sse connection");
+		if (client) {
+			logger.info("unlistening for change");
+			client.query("UNLISTEN change").catch(logger.error);
+			logger.info("releasing client");
+			client.release();
+		}
+	};
+};
+
+const constructChangeChannel = (communityId: string, table: NotifyTables) => {
+	return `change_${communityId}_${table}`;
+};
+
+// bit awkward since we want to read the search params here, but the next-use-sse does not expose the request
+export const GET = (req: NextRequest) => {
+	return createSSEHandler(async (send, close, { onClose }) => {
+		const listen = parseNotifyTables(req.nextUrl.searchParams.getAll("listen"));
+
+		if (!listen?.length) {
+			logger.info("no listen tables, closing sse connection");
+			return handleClose();
+		}
+
+		logger.info("opening sse connection");
+		const [{ user }, community] = await Promise.all([getLoginData(), findCommunityBySlug()]);
+
+		if (!user) {
+			logger.info("no user found, closing sse connection");
+			return handleClose();
+		}
+
+		if (!community) {
+			logger.info("no community found, closing sse connection");
+			return handleClose();
+		}
+
+		const client = await pool.connect();
+
+		listen.forEach(async (table) => {
+			const channelName = constructChangeChannel(community.id, table);
+			await client.query(`LISTEN "${channelName}"`);
+		});
+
+		// handle postgres notifications
+		client.on("notification", async (msg) => {
+			if (!msg.payload) return;
+
+			try {
+				const notification = JSON.parse(msg.payload) as ChangeNotification<NotifyTables>;
+
+				if (!listen.includes(notification.table)) {
+					logger.info("not listening to this table, skipping");
+					return;
+				}
+
+				logger.info({ msg: "notification", notification });
+				send(notification, "change");
+			} catch (err) {
+				logger.error({ msg: "Failed to parse notification:", err });
+			}
+		});
+
+		return handleClose(client);
+	})(req);
+};