Merge pull request #713 from desci-labs/publish-task-q

Improve Publish tracking/recovery

desci-server/nodemon.json

@@ -5,5 +5,5 @@
   "verbose": true,
   "exec": "node -r ts-node/register --inspect=0.0.0.0:9228",
   "delay": 1337,
-  "signal": "SIGTERM"
+  "signal": "SIGINT"
 }

desci-server/package.json

@@ -17,6 +17,7 @@
     "script:fix-data-refs": "debug=* node --inspect=0.0.0.0:9277 --no-warnings --enable-source-maps --loader ts-node/esm ./src/scripts/dataRefDoctor.ts",
     "script:active-users": "debug=* node --no-warnings --enable-source-maps --loader ts-node/esm ./src/scripts/activeUsers.ts",
     "script:invalidate-redis-cache": "debug=* node --no-warnings --enable-source-maps --loader ts-node/esm ./src/scripts/invalidate-redis-cache.ts",
+    "script:fix-publish": "NODE_OPTIONS='--loader ts-node/esm' debug=* node --inspect=0.0.0.0:9277 --no-warnings --enable-source-maps ./src/scripts/fixPublish.ts",
     "script:increase-base-drive-storage": "debug=* node --no-warnings --enable-source-maps --loader ts-node/esm ./src/scripts/increase-base-drive-storage.ts",
     "script:testing": "debug=* node --no-warnings --enable-source-maps --loader ts-node/esm ./src/scripts/testing.ts",
     "script:migrate-draft-trees": "debug=* node --no-warnings --enable-source-maps --loader ts-node/esm ./src/scripts/migrate-draft-trees.ts",

desci-server/prisma/migrations/20241205220126_add_publish_status/migration.sql

@@ -0,0 +1,35 @@
+-- CreateTable
+CREATE TABLE "PublishStatus" (
+    "id" SERIAL NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+    "nodeUuid" TEXT NOT NULL,
+    "version" INTEGER NOT NULL,
+    "versionId" INTEGER,
+    "commitId" TEXT,
+    "ceramicComit" BOOLEAN,
+    "assignDpid" BOOLEAN,
+    "createPdr" BOOLEAN,
+    "fireDeferredEmails" BOOLEAN,
+    "fireNotifications" BOOLEAN,
+    "updateAttestations" BOOLEAN,
+    "transformDraftComments" BOOLEAN,
+    "triggerDoiMint" BOOLEAN,
+
+    CONSTRAINT "PublishStatus_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "PublishStatus_nodeUuid_key" ON "PublishStatus"("nodeUuid");
+
+-- CreateIndex
+CREATE INDEX "PublishStatus_nodeUuid_idx" ON "PublishStatus"("nodeUuid");
+
+-- CreateIndex
+CREATE UNIQUE INDEX "PublishStatus_nodeUuid_version_key" ON "PublishStatus"("nodeUuid", "version");
+
+-- AddForeignKey
+ALTER TABLE "PublishStatus" ADD CONSTRAINT "PublishStatus_versionId_fkey" FOREIGN KEY ("versionId") REFERENCES "NodeVersion"("id") ON DELETE SET NULL ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "PublishStatus" ADD CONSTRAINT "PublishStatus_nodeUuid_fkey" FOREIGN KEY ("nodeUuid") REFERENCES "Node"("uuid") ON DELETE RESTRICT ON UPDATE CASCADE;

desci-server/prisma/migrations/20241206120942_fix_publish_status_unique_constraint/migration.sql

@@ -0,0 +1,2 @@
+-- DropIndex
+DROP INDEX "PublishStatus_nodeUuid_key";

desci-server/prisma/migrations/20241210102946_add_handle_node_version_entry_check/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "PublishStatus" ADD COLUMN     "handleNodeVersionEntry" BOOLEAN;

desci-server/prisma/migrations/20241211184344_add_unique_constraint_on_commit_id/migration.sql

@@ -0,0 +1,13 @@
+/*
+  Warnings:
+
+  - You are about to drop the column `ceramicComit` on the `PublishStatus` table. All the data in the column will be lost.
+  - A unique constraint covering the columns `[commitId]` on the table `PublishStatus` will be added. If there are existing duplicate values, this will fail.
+
+*/
+-- AlterTable
+ALTER TABLE "PublishStatus" DROP COLUMN "ceramicComit",
+ADD COLUMN     "ceramicCommit" BOOLEAN;
+
+-- CreateIndex
+CREATE UNIQUE INDEX "PublishStatus_commitId_key" ON "PublishStatus"("commitId");

desci-server/prisma/migrations/20241211203631_add_manifest_cid_publish_status_table/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "PublishStatus" ADD COLUMN     "manifestCid" TEXT;

desci-server/prisma/schema.prisma

@@ -53,6 +53,7 @@ model Node {
   DeferredEmails       DeferredEmails[]
   UserNotifications    UserNotifications[]
   Annotation           Annotation[]
+  PublishStatus        PublishStatus[]
   ExternalPublications ExternalPublications[]
 
   @@index([ownerId])
@@ -71,6 +72,7 @@ model NodeVersion {
   node                Node?                 @relation(fields: [nodeId], references: [id])
   PublicDataReference PublicDataReference[]
   DataReference       DataReference[]
+  PublishStatus       PublishStatus[]
 }
 
 model InteractionLog {
@@ -937,6 +939,31 @@ model UserNotifications {
   user      User             @relation(fields: [userId], references: [id])
 }
 
+model PublishStatus {
+  id                     Int          @id @default(autoincrement())
+  createdAt              DateTime     @default(now())
+  updatedAt              DateTime     @updatedAt
+  nodeUuid               String
+  version                Int
+  versionId              Int?
+  commitId               String?      @unique
+  manifestCid            String?
+  ceramicCommit          Boolean?
+  assignDpid             Boolean?
+  handleNodeVersionEntry Boolean?
+  createPdr              Boolean?
+  fireDeferredEmails     Boolean?
+  fireNotifications      Boolean?
+  updateAttestations     Boolean?
+  transformDraftComments Boolean?
+  triggerDoiMint         Boolean?
+  nodeVersion            NodeVersion? @relation(fields: [versionId], references: [id])
+  node                   Node         @relation(fields: [nodeUuid], references: [uuid])
+
+  @@unique([nodeUuid, version])
+  @@index([nodeUuid])
+}
+
 model ExternalPublications {
   id          Int      @id @default(autoincrement())
   uuid        String

desci-server/src/controllers/admin/debug.ts

@@ -1,10 +1,11 @@
-import { DataType, Node, NodeVersion, PublicDataReference } from '@prisma/client';
+import { DataType, Node, NodeVersion, PublicDataReference, PublishStatus } from '@prisma/client';
 import { Request, Response } from 'express';
 
 import { prisma } from '../../client.js';
 import { logger as parentLogger } from '../../logger.js';
 import { directStreamLookup, RawStream } from '../../services/ceramic.js';
 import { getAliasRegistry, getHotWallet } from '../../services/chain.js';
+import { PublishServices } from '../../services/PublishServices.js';
 import { _getIndexedResearchObjects, getIndexedResearchObjects, IndexedResearchObject } from '../../theGraph.js';
 import { ensureUuidEndsWithDot, hexToCid } from '../../utils.js';
 
@@ -127,6 +128,7 @@ type NodeDebugReport =
       database: any;
       indexer: any;
       migration: any;
+      publishStatus: DebugPublishStatusResponse;
     }
   | { hasError: true; reason: string };
 
@@ -158,6 +160,7 @@ const debugNode = async (uuid: string): Promise<NodeDebugReport> => {
   const shouldBeIndexed = database.nVersions > 0 || stream.nVersions > 0;
   const indexer = await debugIndexer(uuid, shouldBeIndexed);
   const migration = await debugMigration(uuid, stream);
+  const publishStatus = await debugPublishStatus(uuid);
 
   const nVersionsAgree = new Set([database.nVersions ?? 0, stream.nVersions ?? 0, indexer.nVersions ?? 0]).size === 1;
 
@@ -174,6 +177,7 @@ const debugNode = async (uuid: string): Promise<NodeDebugReport> => {
     database,
     indexer,
     migration,
+    publishStatus,
   };
 };
 
@@ -235,6 +239,32 @@ const debugStream = async (stream?: string): Promise<DebugStreamResponse> => {
   };
 };
 
+type DebugPublishStatusResponse =
+  | {
+      present: boolean;
+      error: false;
+      publishStatus: PublishStatus[];
+    }
+  | {
+      present: boolean;
+      error: true;
+      name: string;
+      message: string;
+      stack: unknown;
+    };
+/*
+ ** PublishStatus shows the steps taken during the publish process and which have suceeded/failed/uninitialized
+ */
+const debugPublishStatus = async (nodeUuid: string): Promise<DebugPublishStatusResponse> => {
+  try {
+    const publishStatus = await PublishServices.getPublishStatusForNode(ensureUuidEndsWithDot(nodeUuid));
+    return { present: true, publishStatus, error: false };
+  } catch (e) {
+    const err = e as Error;
+    return { present: true, error: true, name: err.name, message: err.message, stack: err.stack };
+  }
+};
+
 const debugDpid = async (dpidAlias?: number) => {
   if (!dpidAlias) return { present: false, error: false };
 

desci-server/src/controllers/admin/publish/resumePublish.ts

@@ -0,0 +1,59 @@
+import { User } from '@prisma/client';
+import { Request, Response } from 'express';
+import { z } from 'zod';
+
+import { logger as parentLogger } from '../../../logger.js';
+import { publishSequencer } from '../../../services/PublishServices.js';
+
+const ResumePublishSchema = z
+  .object({
+    publishStatusId: z.number().optional(),
+    commitId: z.string().optional(),
+    nodeUuid: z.string().optional(),
+    version: z.number().optional(),
+  })
+  .refine((data) => !!(data.publishStatusId || data.commitId || (data.nodeUuid && data.version)), {
+    message: 'Must provide either publishStatusId, commitId, or both nodeUuid and version',
+  });
+
+export interface AuthenticatedRequest extends Request {
+  user: User;
+}
+
+export interface ErrorResponse {
+  allStepsSucceeded: false;
+  error: string;
+  details?: z.ZodIssue[] | string;
+}
+
+export const resumePublish = async (
+  req: AuthenticatedRequest & { body: z.infer<typeof ResumePublishSchema> },
+  res: Response<{ allStepsSucceeded: boolean } | ErrorResponse>,
+) => {
+  const logger = parentLogger.child({
+    module: 'Admin:Publish::resumePublish',
+    userId: req.user?.id,
+    body: req.body,
+  });
+  try {
+    const args = ResumePublishSchema.parse(req.body);
+    logger.debug({ args }, 'Resuming publish');
+    const success = await publishSequencer(args);
+
+    return res.status(success ? 200 : 500).json({ allStepsSucceeded: success });
+  } catch (error) {
+    if (error instanceof z.ZodError) {
+      return res.status(400).json({
+        allStepsSucceeded: false,
+        error: 'Invalid request parameters',
+        details: error.errors,
+      });
+    }
+
+    logger.error({ error }, 'Error resuming publish');
+    return res.status(500).json({
+      allStepsSucceeded: false,
+      error: error instanceof Error ? error.message : String(error),
+    });
+  }
+};

desci-server/src/controllers/nodes/contributions/prepubEmail.ts

@@ -3,7 +3,7 @@ import type { Request, Response } from 'express';
 
 import { prisma } from '../../../client.js';
 import { logger as parentLogger } from '../../../logger.js';
-import { publishServices } from '../../../services/PublishServices.js';
+import { PublishServices } from '../../../services/PublishServices.js';
 import { CidString } from '../../../services/Thumbnails.js';
 import { ensureUuidEndsWithDot } from '../../../utils.js';
 
@@ -70,7 +70,7 @@ export const emailPublishPackage = async (
     // if (!distPdfEntry) return res.status(404).json({ ok: false, error: 'Distribution PDF not found' });
 
     // Fire off email
-    await publishServices.sendVersionUpdateEmailToAllContributors({
+    await PublishServices.sendVersionUpdateEmailToAllContributors({
       node,
       manuscriptCid: prepubDistPdfCid,
       ownerOnly: !emailAllContributors,