Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions package.json
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
"start": "node dist/index.js",
"start:publish-reconcile": "node dist/cron/publishReconcile.js",
"dev:publish-reconcile": "tsx watch src/cron/publishReconcile.ts",
"start:grace-purge": "node dist/cron/gracePurge.js",
"dev:grace-purge": "tsx watch src/cron/gracePurge.ts",
"clean": "rm -rf dist",
"format": "prettier --write .",
"lint": "eslint src/**/*.ts",
Expand Down
12 changes: 12 additions & 0 deletions src/config/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,19 @@ export const config: ServerConfig = {
interval: process.env.PUBLISH_RECONCILE_INTERVAL || "15 minutes",
batchSize: parseInt(process.env.PUBLISH_RECONCILE_BATCH || "500"),
},
deleteGrace: {
windowMs: parseInt(process.env.DELETE_GRACE_WINDOW_MS || "2592000000"), // 30 days
interval: process.env.DELETE_GRACE_INTERVAL || "1 hour",
batchSize: parseInt(process.env.DELETE_GRACE_BATCH || "200"),
},
agenda: {
concurrency: parseInt(process.env.AGENDA_DEFAULT_CONCURRENCY || "1"),
},
gate: {
url: process.env.GATE_URL,
did: process.env.GATE_DID,
},
webhook: {
apiKey: process.env.COLLAB_WEBHOOK_API_KEY,
},
};
26 changes: 26 additions & 0 deletions src/cron/gracePurge.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
import { agenda } from "./agenda.js";
import { databaseService } from "../database/index.js";
import gracePurgeJob from "./jobs/gracePurgeJob.js";

async function graceful() {
await agenda.stop();
await databaseService.disconnect();
process.exit(0);
}

(async () => {
try {
// The job queries mongoose models, so this worker needs its own DB connection
// (Agenda's Mongo connection only backs the jobs collection).
await databaseService.connect();
await agenda.start();
await gracePurgeJob.setupJob();
console.log("grace-purge worker started");
} catch (err) {
console.error(err);
await graceful();
}
})();

process.on("SIGTERM", graceful);
process.on("SIGINT", graceful);
25 changes: 25 additions & 0 deletions src/cron/jobs/gracePurgeJob.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
import type { Job } from "agenda";
import { agenda } from "../agenda.js";
import { config } from "../../config/index.js";
import { mongodbStore } from "../../services/mongodb-store.js";

const jobName = "DELETE_GRACE_PURGE";

export async function jobDefinition(job: Job, done: (err?: Error) => void) {
try {
const cutoff = Date.now() - config.deleteGrace.windowMs;
const purged = await mongodbStore.purgeTombstonedBefore(cutoff, config.deleteGrace.batchSize);
console.log(`[${jobName}] purged=${purged.length}`);
done();
} catch (err: any) {
console.error(`[${jobName}] ${err?.message ?? err}`);
done(err);
}
}

async function setupJob() {
agenda.define(jobName, jobDefinition);
await agenda.every(config.deleteGrace.interval, jobName);
}

export default { setupJob };
20 changes: 20 additions & 0 deletions src/database/models/document-edit-epoch.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
import mongoose, { Schema, Document as MongooseDocument } from "mongoose";

// Per-document edit-admission floor. A gp-actor editUcan whose `epoch` fact is below this
// is refused at JOIN (offline revocation — replaces the runtime edit-bound poll). Bumped by
// the rotation / evict path. See docs/architecture/gp-semaphore.md.
interface IDocumentEditEpoch extends MongooseDocument {
_id: string; // documentId
minEditEpoch: number;
}

const DocumentEditEpochSchema = new Schema<IDocumentEditEpoch>({
_id: { type: String, required: true },
minEditEpoch: { type: Number, required: true, default: 0 },
});

export const DocumentEditEpochModel = mongoose.model<IDocumentEditEpoch>(
"DocumentEditEpoch",
DocumentEditEpochSchema
);
export type { IDocumentEditEpoch };
6 changes: 6 additions & 0 deletions src/database/models/document-meta.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@ interface IDocumentMeta extends MongooseDocument {
title: string | null; // roomKey-encrypted
updatedAt: number;
isPublished: boolean; // set by the publish reconciler once the doc is on-chain
tombstonedAt: number | null; // set by the DeletedFile webhook; grace window runs from here
tombstoneReason: string | null;
}

const DocumentMetaSchema = new Schema<IDocumentMeta>({
Expand All @@ -22,13 +24,17 @@ const DocumentMetaSchema = new Schema<IDocumentMeta>({
title: { type: String, default: null },
updatedAt: { type: Number, required: true },
isPublished: { type: Boolean, default: false },
tombstonedAt: { type: Number, default: null },
tombstoneReason: { type: String, default: null },
});

// Discovery: "all docs bound to this identity / portal owner" for list-my-documents.
// Compound with isPublished so both the (unpublished-only) list read and the
// reconciler's candidate scan are index-served.
DocumentMetaSchema.index({ ownerIdentityDid: 1, isPublished: 1 }, { background: true });
DocumentMetaSchema.index({ ownerDid: 1 }, { background: true });
// Grace-window sweep: docs eligible for irreversible purge (see docs/architecture/edit-permission.md).
DocumentMetaSchema.index({ tombstonedAt: 1 }, { background: true });

export const DocumentMetaModel = mongoose.model<IDocumentMeta>("DocumentMeta", DocumentMetaSchema);
export type { IDocumentMeta };
25 changes: 25 additions & 0 deletions src/database/models/document-mirror.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
import mongoose, { Schema, Document as MongooseDocument } from "mongoose";

// The VIEW plane: fileKey-encrypted full-state snapshots (published-artifact shape) that
// viewers read load-on-open. Separate from DocumentUpdate (roomKey durability lane): different
// key, open read, epoch-keyed. One row per (documentId, fileKeyEpoch), keep-latest.
interface IDocumentMirror extends MongooseDocument {
documentId: string;
fileKeyEpoch: number;
data: string;
createdAt: number;
authorSessionDid: string;
}

const DocumentMirrorSchema = new Schema<IDocumentMirror>({
documentId: { type: String, required: true },
fileKeyEpoch: { type: Number, required: true },
data: { type: String, required: true },
createdAt: { type: Number, required: true },
authorSessionDid: { type: String, required: true },
});

DocumentMirrorSchema.index({ documentId: 1, fileKeyEpoch: 1 }, { unique: true, background: true });

export const DocumentMirrorModel = mongoose.model<IDocumentMirror>("DocumentMirror", DocumentMirrorSchema);
export type { IDocumentMirror };
2 changes: 2 additions & 0 deletions src/database/models/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,3 +3,5 @@ export { DocumentCommitModel, type IDocumentCommit } from "./document-commit";
export { SessionModel, type ISession } from "./session";
export { CounterModel, type ICounter } from "./counter";
export { DocumentMetaModel, type IDocumentMeta } from "./document-meta";
export { DocumentMirrorModel, type IDocumentMirror } from "./document-mirror";
export { DocumentEditEpochModel, type IDocumentEditEpoch } from "./document-edit-epoch";
2 changes: 2 additions & 0 deletions src/database/models/session.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ export interface ISession extends MongooseDocument {
ownerIdentityDid: string | null;
portalAddress: string | null;
collabJoinEnabled?: boolean;
workspaceEditEnabled?: boolean;
}

const SessionSchema = new Schema({
Expand All @@ -26,6 +27,7 @@ const SessionSchema = new Schema({
ownerIdentityDid: { type: String, default: null },
portalAddress: { type: String, default: null },
collabJoinEnabled: { type: Boolean }, // intentionally no default
workspaceEditEnabled: { type: Boolean }, // intentionally no default
});

SessionSchema.index({ documentId: 1, createdAt: 1, sessionDid: 1 });
Expand Down
75 changes: 74 additions & 1 deletion src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,18 +7,27 @@ import compression from "compression";
import { createServer } from "http";
import { config } from "./config";
import { authService } from "./services/auth";
import { registerEventHandlers } from "./services/socket-handlers";
import { registerEventHandlers, getRoomName } from "./services/socket-handlers";
import { authMiddleware } from "./services/auth-middleware";
import { sessionManager } from "./services/session-manager";
import { mongodbStore } from "./services/mongodb-store";
import { createRedisAdapter } from "./redis";
import { databaseService } from "./database";
import {
createCollabJoinEnabledHandler,
createWorkspaceEditTierHandler,
createEvictEditActorsHandler,
createEvictWorkspaceMemberHandler,
createListMyDocumentsHandler,
createDeleteDocumentHandler,
createMirrorReadHandler,
createShareContextHandler,
} from "./services/owner-op-routes";
import { createRotateSessionHandler } from "./services/rotate-route";
import { rotationCoordinator } from "./services/rotation-coordinator";
import { editBoundCache } from "./services/gate-epoch";
import { createFlushHandler } from "./services/flush-route";
import { createDeletedFileWebhookHandler } from "./services/deleted-file-webhook";
import { createLightNode } from "@waku/sdk";
import protobuf from "protobufjs";
import { generateKeyPairFromSeed } from "@libp2p/crypto/keys";
Expand Down Expand Up @@ -95,6 +104,58 @@ class CollaborationServer {
"/documents/:documentId/collab-join-enabled",
createCollabJoinEnabledHandler({ authService, sessionManager }, this.io)
);
this.app.post(
"/documents/:documentId/workspace-edit-tier",
createWorkspaceEditTierHandler({ authService, sessionManager }, this.io)
);
this.app.post(
"/documents/:documentId/evict-edit-actors",
createEvictEditActorsHandler({ authService, sessionManager, editBoundCache, mongodbStore }, this.io)
);
this.app.post(
"/workspaces/:portalAddress/evict-member",
createEvictWorkspaceMemberHandler({ authService, sessionManager }, this.io)
);
this.app.post(
"/documents/:documentId/rotate-session",
createRotateSessionHandler({
authService, sessionManager, mongodbStore, rotationCoordinator,
terminateOldSession: async (documentId, sessionDid, appType) => {
const room = getRoomName(documentId, sessionDid);
const sockets = await this.io!.in(room).fetchSockets();
// Laggard sockets stay AUTHED: their next write must reach createUpdate and get the
// D-11 SESSION_TERMINATED 409 — that ack is their self-heal signal.
// De-authing here would surface a 401 first and strand them frozen.
// See docs/architecture/gp-semaphore.md.
for (const s of sockets) { s.leave(room); }
await sessionManager.deactivateSession(documentId, sessionDid);
await sessionManager.terminateSession(documentId, sessionDid, appType);
},
}, this.io)
);
this.app.get(
"/documents/:documentId/mirror",
createMirrorReadHandler({ mongodbStore })
);
this.app.get(
"/documents/:documentId/share-context",
createShareContextHandler({ mongodbStore, sessionManager })
);
this.app.post(
"/webhooks/file-deleted",
createDeletedFileWebhookHandler({
mongodbStore,
onTombstoned: async (documentId) => {
const sessions = await sessionManager.getNonTerminatedSessionsForDocument(documentId);
for (const s of sessions) {
const room = getRoomName(documentId, s.sessionDid);
this.io!.to(room).emit("/session/terminated", { roomId: documentId });
for (const sock of await this.io!.in(room).fetchSockets()) sock.leave(room);
await sessionManager.terminateSession(documentId, s.sessionDid, s.appType ?? "ddoc");
}
},
})
);
this.app.post("/flush", createFlushHandler({ authService, mongodbStore }));
this.app.post("/list-my-documents", createListMyDocumentsHandler({ authService, mongodbStore }));
this.app.delete(
Expand Down Expand Up @@ -124,6 +185,18 @@ class CollaborationServer {

async start() {
try {
if (!config.gate.did) {
console.warn(
"[startup] GATE_DID is not set — GP (private/group) live editing is DISABLED; only owner/workspace/public rails admit writes."
);
} else if (!config.gate.url) {
// GATE_DID without GATE_URL: verifyEditUcan admits GP joins but the edit-bound cache
// can never reach the gate, so every revocation (demote/revoke) is a permanent silent no-op.
throw new Error(
"[startup] GATE_DID is set but GATE_URL is not — GP edit revocation would never take effect. Set GATE_URL to the gate origin, or unset GATE_DID to disable GP editing."
);
}

// Initialize database connection
await databaseService.connect();

Expand Down
Loading
Loading