[feat] Unify async DKG Publisher setup, management, and runtime fixes

README.md

@@ -54,6 +54,7 @@ Before running the project setup, you'll need to configure the following environ
   - **Testnet** (safe testing with mock tokens): `https://v6-pegasus-node-02.origin-trail.network:8900`
   - **Mainnet** (production DKG interactions): `https://positron.origin-trail.network`
   - **Local development**: `http://localhost:8900` (default)
+- **`DKG_NODE_CUSTOM_RPC`**: Optional custom blockchain RPC URL. Leave unset to use the default RPC behavior.
 - **`PORT`**: Server port (default: `9200`)
 - **`EXPO_PUBLIC_APP_URL`**: Public app URL (default: `http://localhost:9200`)
 - **`EXPO_PUBLIC_MCP_URL`**: MCP server URL (default: `http://localhost:9200`)
@@ -71,8 +72,21 @@ The setup script will:
 - Prompt for required environment variables
 - Create `.env` and `.env.development.local` files
 - Set up the SQLite database with migrations
+- Optionally enable async publishing and provision the Publisher database using the Engine-derived MySQL password
 - Create an admin user (username: `admin`, password: `admin123`)
 
+If you enable async publishing during setup, the Agent writes the consolidated Publisher settings into `apps/agent/.env`:
+
+- `ASYNC_PUBLISHING_ENABLED=true`
+- `DKGP_DATABASE_URL`
+- `REDIS_URL`
+
+The setup stores `DKG_PUBLISH_WALLET` in `apps/agent/.env` without a `0x`
+prefix for compatibility with existing node env files, and Publisher wallet
+records are seeded into MySQL using the same bare 64-hex format.
+
+The Agent server only loads `@dkg/plugin-dkg-publisher` when `ASYNC_PUBLISHING_ENABLED=true`.
+
 ### 4. Start Development
 
 ```bash

apps/agent/README.md

@@ -23,6 +23,11 @@
    npm run script:setup
    ```
 
+   The setup flow can also enable async publishing. When enabled, it reuses the
+   Engine MySQL password from `dkg-engine/current/.env`, writes the consolidated
+   Publisher settings into `apps/agent/.env`, and the Agent loads the Publisher
+   plugin only when `ASYNC_PUBLISHING_ENABLED=true`.
+
 Now you can run the app in development mode using:
 
 ```bash

apps/agent/env.d.ts

@@ -14,6 +14,15 @@ declare global {
       DKG_PUBLISH_WALLET: string;
       DKG_BLOCKCHAIN: string;
       DKG_OTNODE_URL: string;
+      DKG_NODE_CUSTOM_RPC?: string;
+      ASYNC_PUBLISHING_ENABLED?: string;
+      DKGP_DATABASE_URL?: string;
+      REDIS_URL?: string;
+      WORKER_COUNT?: string;
+      POLL_FREQUENCY?: string;
+      STORAGE_TYPE?: string;
+      STORAGE_PATH?: string;
+      STORAGE_BASE_URL?: string;
       SMTP_HOST: string;
       SMTP_PORT: string;
       SMTP_USER: string;

apps/agent/package.json

@@ -17,6 +17,7 @@
     "dev:server": "tsup src/server/*.ts --format esm,cjs --watch --onSuccess=\"node dist/index.js --dev\"",
     "drizzle:studio": "drizzle-kit studio",
     "script:setup": "node dist/scripts/setup.js",
+    "script:publisher": "node dist/scripts/publisher.js",
     "script:createUser": "node dist/scripts/createUser.js",
     "script:createToken": "node dist/scripts/createToken.js",
     "test": "PLAYWRIGHT_JUNIT_OUTPUT_NAME=DKG_Node_UI_Tests.xml npx playwright test spec/testUI.spec.js --grep '@gh_actions' --reporter=list,html,junit",
@@ -32,6 +33,7 @@
   "dependencies": {
     "@dkg/expo-forcegraph": "^0.0.0",
     "@dkg/plugin-dkg-essentials": "^0.0.3",
+    "@dkg/plugin-dkg-publisher": "^1.0.0",
     "@dkg/plugin-example": "^0.0.3",
     "@dkg/plugin-oauth": "^0.0.2",
     "@dkg/plugin-swagger": "^0.0.2",

apps/agent/src/server/helpers.ts

@@ -34,6 +34,10 @@ export async function createFileWithContent(filePath: string, content: string) {
   }
 }
 
+export async function writeFileWithContent(filePath: string, content: string) {
+  await fs.writeFile(filePath, content, { encoding: "utf8" });
+}
+
 export function configEnv() {
   dotenv.config();
   if (process.argv.includes("--dev")) {

apps/agent/src/server/index.ts

@@ -1,109 +1,94 @@
 import path from "path";
-import { createPluginServer, defaultPlugin } from "@dkg/plugins";
+import {
+  createPluginServer,
+  defaultPlugin,
+  type DkgPlugin,
+} from "@dkg/plugins";
 import { authorized, createOAuthPlugin } from "@dkg/plugin-oauth";
 import dkgEssentialsPlugin from "@dkg/plugin-dkg-essentials";
 import createFsBlobStorage from "@dkg/plugin-dkg-essentials/createFsBlobStorage";
 import examplePlugin from "@dkg/plugin-example";
 import swaggerPlugin from "@dkg/plugin-swagger";
-//@ts-expect-error No types for dkg.js ...
+// @ts-expect-error No types for dkg.js ...
 import DKG from "dkg.js";
 import { eq } from "drizzle-orm";
+import { getTestMessageUrl } from "nodemailer";
 
 import { userCredentialsSchema } from "@/shared/auth";
 import { processStreamingCompletion } from "@/shared/chat";
 import { verify } from "@node-rs/argon2";
 
-import { configDatabase, configEnv } from "./helpers";
-import webInterfacePlugin from "./webInterfacePlugin";
 import createAccountManagementPlugin from "./accountManagementPlugin";
 import {
-  users,
-  SqliteOAuthStorageProvider,
   SqliteAccountManagementProvider,
+  SqliteOAuthStorageProvider,
+  users,
 } from "./database/sqlite";
+import { configDatabase, configEnv } from "./helpers";
 import mailer from "./mailer";
-import { getTestMessageUrl } from "nodemailer";
-
-configEnv();
-const db = configDatabase();
-
-const version = "1.0.0";
-
-const { oauthPlugin, openapiSecurityScheme } = createOAuthPlugin({
-  storage: new SqliteOAuthStorageProvider(db),
-  issuerUrl: new URL(process.env.EXPO_PUBLIC_MCP_URL),
-  scopesSupported: [
-    "mcp",
-    "llm",
-    "scope123",
-    "blob",
-    "epcis.read",
-    "epcis.write",
-  ],
-  loginPageUrl: new URL(process.env.EXPO_PUBLIC_APP_URL + "/login"),
-  schema: userCredentialsSchema,
-  async login(credentials) {
-    const user = await db
-      .select()
-      .from(users)
-      .where(eq(users.email, credentials.email))
-      .then((r) => r.at(0));
-    if (!user) throw new Error("Invalid credentials");
+import webInterfacePlugin from "./webInterfacePlugin";
 
-    const isValid = await verify(user.password, credentials.password);
-    if (!isValid) throw new Error("Invalid credentials");
+async function main() {
+  configEnv();
+  const db = configDatabase();
+  const version = "1.0.0";
 
-    return { scopes: user.scope.split(" "), extra: { userId: user.id } };
-  },
-});
+  const { oauthPlugin, openapiSecurityScheme } = createOAuthPlugin({
+    storage: new SqliteOAuthStorageProvider(db),
+    issuerUrl: new URL(process.env.EXPO_PUBLIC_MCP_URL),
+    scopesSupported: [
+      "mcp",
+      "llm",
+      "scope123",
+      "blob",
+      "epcis.read",
+      "epcis.write",
+    ],
+    loginPageUrl: new URL(process.env.EXPO_PUBLIC_APP_URL + "/login"),
+    schema: userCredentialsSchema,
+    async login(credentials) {
+      const user = await db
+        .select()
+        .from(users)
+        .where(eq(users.email, credentials.email))
+        .then((results) => results.at(0));
+      if (!user) throw new Error("Invalid credentials");
 
-const accountManagementPlugin = createAccountManagementPlugin({
-  provider: new SqliteAccountManagementProvider(db),
-  async sendMail(toEmail, code) {
-    const m = await mailer();
-    if (!m) throw new Error("No SMTP transport available");
+      const isValid = await verify(user.password, credentials.password);
+      if (!isValid) throw new Error("Invalid credentials");
 
-    await m
-      .sendMail({
-        to: toEmail,
-        subject: "Password reset request | DKG Node",
-        text:
-          `Your password reset code is ${code}.` +
-          `Link: ${process.env.EXPO_PUBLIC_APP_URL}/password-reset?code=${code}`,
-        html:
-          `<p>Your password reset code is <strong>${code}</strong>.</p>` +
-          `<p>Please click <a href="${process.env.EXPO_PUBLIC_APP_URL}/password-reset?code=${code}">here</a> to reset your password.</p>`,
-      })
-      .then((info) => {
-        console.debug(info);
-        console.debug(getTestMessageUrl(info));
-      });
-  },
-});
+      return { scopes: user.scope.split(" "), extra: { userId: user.id } };
+    },
+  });
 
-const blobStorage = createFsBlobStorage(path.join(__dirname, "../data"));
+  const accountManagementPlugin = createAccountManagementPlugin({
+    provider: new SqliteAccountManagementProvider(db),
+    async sendMail(toEmail, code) {
+      const transport = await mailer();
+      if (!transport) throw new Error("No SMTP transport available");
 
-const otnodeUrl = new URL(process.env.DKG_OTNODE_URL);
+      await transport
+        .sendMail({
+          to: toEmail,
+          subject: "Password reset request | DKG Node",
+          text:
+            `Your password reset code is ${code}.` +
+            `Link: ${process.env.EXPO_PUBLIC_APP_URL}/password-reset?code=${code}`,
+          html:
+            `<p>Your password reset code is <strong>${code}</strong>.</p>` +
+            `<p>Please click <a href="${process.env.EXPO_PUBLIC_APP_URL}/password-reset?code=${code}">here</a> to reset your password.</p>`,
+        })
+        .then((info) => {
+          console.debug(info);
+          console.debug(getTestMessageUrl(info));
+        });
+    },
+  });
 
-const app = createPluginServer({
-  name: "DKG API",
-  version,
-  context: {
-    blob: blobStorage,
-    dkg: new DKG({
-      endpoint: `${otnodeUrl.protocol}//${otnodeUrl.hostname}`,
-      port: otnodeUrl.port || "8900",
-      blockchain: {
-        name: process.env.DKG_BLOCKCHAIN,
-        privateKey: process.env.DKG_PUBLISH_WALLET,
-      },
-      maxNumberOfRetries: 300,
-      frequency: 2,
-      contentType: "all",
-      nodeApiVersion: "/v1",
-    }),
-  },
-  plugins: [
+  const blobStorage = createFsBlobStorage(path.join(__dirname, "../data"));
+  const otnodeUrl = new URL(process.env.DKG_OTNODE_URL);
+  const dkgCustomRpc = process.env.DKG_NODE_CUSTOM_RPC?.trim();
+  const plugins: DkgPlugin[] = [
     defaultPlugin,
     oauthPlugin,
     (_, __, api) => {
@@ -119,7 +104,6 @@ const app = createPluginServer({
       api.use("/change-password", authorized([]));
       api.use("/profile", authorized([]));
     },
-    // Streaming LLM middleware — intercepts SSE requests before Expo Router
     (_, __, api) => {
       api.post("/llm", (req, res, next) => {
         if (!req.headers.accept?.includes("text/event-stream")) return next();
@@ -128,8 +112,18 @@ const app = createPluginServer({
     },
     accountManagementPlugin,
     dkgEssentialsPlugin,
+  ];
+
+  if (process.env.ASYNC_PUBLISHING_ENABLED === "true") {
+    const { default: dkgPublisherPlugin } = await import(
+      "@dkg/plugin-dkg-publisher"
+    );
+    plugins.push(dkgPublisherPlugin);
+  }
+
+  plugins.push(
     examplePlugin.withNamespace("protected", {
-      middlewares: [authorized(["scope123"])], // Allow only users with the "scope123" scope
+      middlewares: [authorized(["scope123"])],
     }),
     swaggerPlugin({
       version,
@@ -145,28 +139,55 @@ const app = createPluginServer({
       ],
     }),
     webInterfacePlugin(path.join(__dirname, "./app")),
-  ],
-});
-
-const port = process.env.PORT || 9200;
-const server = app.listen(port, (err) => {
-  if (err) {
-    console.error(err);
-    process.exit(1);
-  }
-  console.log(`Server running at http://localhost:${port}/`);
+  );
 
-  process.on("SIGINT", () => {
-    server.close();
-    process.exit(0);
+  const app = createPluginServer({
+    name: "DKG API",
+    version,
+    context: {
+      blob: blobStorage,
+      dkg: new DKG({
+        endpoint: `${otnodeUrl.protocol}//${otnodeUrl.hostname}`,
+        port: otnodeUrl.port || "8900",
+        blockchain: {
+          name: process.env.DKG_BLOCKCHAIN,
+          privateKey: process.env.DKG_PUBLISH_WALLET,
+          ...(dkgCustomRpc && { rpc: dkgCustomRpc }),
+        },
+        maxNumberOfRetries: 300,
+        frequency: 2,
+        contentType: "all",
+        nodeApiVersion: "/v1",
+      }),
+    },
+    plugins,
   });
-  process.on("SIGTERM", () => {
-    server.close((err) => {
-      if (err) {
-        console.error(err);
-        process.exit(1);
-      }
+
+  const port = process.env.PORT || 9200;
+  const server = app.listen(port, (error) => {
+    if (error) {
+      console.error(error);
+      process.exit(1);
+    }
+    console.log(`Server running at http://localhost:${port}/`);
+
+    process.on("SIGINT", () => {
+      server.close();
       process.exit(0);
     });
+    process.on("SIGTERM", () => {
+      server.close((closeError) => {
+        if (closeError) {
+          console.error(closeError);
+          process.exit(1);
+        }
+        process.exit(0);
+      });
+    });
   });
+}
+
+main().catch((error) => {
+  console.error(error);
+  process.exit(1);
 });