From f462b124e678ce656da59c027760dd07efc34063 Mon Sep 17 00:00:00 2001 From: Sacha Al Himdani Date: Mon, 15 Jun 2026 09:13:31 +0200 Subject: [PATCH] Batch signature and approval notifications via debounced worker Replace the immediate per-document approval email and the manual "send signing notifications" action with a single debounced worker that batches pending requests per recipient and organization. The worker (go.gearno.de/kit/worker) polls on an interval (default 5m) and claims one (organization, recipient) group at a time, sending one consolidated signing email and/or one approval email per recipient/org that lists every document awaiting their signature or approval. The claim is a conditional UPDATE that doubles as concurrency-safe dedup, so several workers never email the same group twice. Each request is notified once it has been pending past the debounce delay (default 15m), then reminded at 1x, 2x and 3x the reminder interval (default 1 day) after the previous email, after which it stops. New last_notified_at and notification_count columns on signatures and approval decisions drive the debounce, the widening reminder cadence and the four-email cap. Email copy lists each document with its title, type and a deep link to the employee page. Removed the inline approval-on-publish email, the SendSigningNotifications service method/mutation/MCP tool, its IAM action, and the related console UI and n8n operation. Signed-off-by: Sacha Al Himdani --- .env.example | 5 + apps/console/src/hooks/graph/DocumentGraph.ts | 23 - .../organizations/documents/DocumentsPage.tsx | 25 - .../_components/CreateDocumentDialog.tsx | 1 - .../documents/_components/DocumentList.tsx | 13 +- e2e/console/tracker_pattern_test.go | 1 + packages/emails/emails.go | 36 +- packages/emails/src/DocumentApproval.tsx | 14 +- packages/emails/src/DocumentSigning.tsx | 12 +- .../emails/templates/document-approval.txt | 6 +- .../emails/templates/document-signing.txt | 6 +- .../nodes/Probo/actions/document/index.ts | 9 - .../sendSigningNotifications.operation.ts | 55 -- pkg/bootstrap/builder.go | 5 + pkg/bootstrap/builder_test.go | 9 + pkg/coredata/document_type.go | 25 + pkg/coredata/document_version.go | 52 ++ .../document_version_approval_decision.go | 214 ++++++- .../document_version_approval_quorum.go | 41 ++ pkg/coredata/document_version_signature.go | 231 +++++++- pkg/coredata/membership_profile.go | 76 --- pkg/coredata/migrations/20260612T150000Z.sql | 31 + pkg/probo/actions.go | 19 +- pkg/probo/document_approval_service.go | 103 +--- pkg/probo/document_notification_worker.go | 549 ++++++++++++++++++ pkg/probo/document_service.go | 100 ---- pkg/probod/aliases.go | 12 +- pkg/probod/probod.go | 40 ++ pkg/probodconfig/notifications_config.go | 20 +- .../api/console/v1/document_resolvers.go | 17 - .../api/console/v1/graphql/document.graphql | 11 - pkg/server/api/mcp/v1/schema.resolvers.go | 18 - pkg/server/api/mcp/v1/specification.yaml | 26 - 33 files changed, 1281 insertions(+), 524 deletions(-) delete mode 100644 packages/n8n-node/nodes/Probo/actions/document/sendSigningNotifications.operation.ts create mode 100644 pkg/coredata/migrations/20260612T150000Z.sql create mode 100644 pkg/probo/document_notification_worker.go diff --git a/.env.example b/.env.example index d4ee7e044..69cb8d06b 100644 --- a/.env.example +++ b/.env.example @@ -79,6 +79,11 @@ # THIRD_PARTY_VETTING_STALE_AFTER=1500 # THIRD_PARTY_VETTING_MAX_CONCURRENCY=1 +# ── Document signature/approval notifications (seconds) ─────────────── +# DOCUMENT_NOTIFICATION_INTERVAL=300 +# DOCUMENT_NOTIFICATION_DEBOUNCE_DELAY=900 +# DOCUMENT_NOTIFICATION_REMINDER_INTERVAL=86400 + # ── OIDC sign-in providers ──────────────────────────────────────────── # AUTH_GOOGLE_CLIENT_ID= # AUTH_GOOGLE_CLIENT_SECRET= diff --git a/apps/console/src/hooks/graph/DocumentGraph.ts b/apps/console/src/hooks/graph/DocumentGraph.ts index c5938c0b4..03b887080 100644 --- a/apps/console/src/hooks/graph/DocumentGraph.ts +++ b/apps/console/src/hooks/graph/DocumentGraph.ts @@ -17,7 +17,6 @@ import { graphql } from "relay-runtime"; import type { DocumentGraphBulkExportDocumentsMutation } from "#/__generated__/core/DocumentGraphBulkExportDocumentsMutation.graphql"; import type { DocumentGraphDeleteMutation } from "#/__generated__/core/DocumentGraphDeleteMutation.graphql"; -import type { DocumentGraphSendSigningNotificationsMutation } from "#/__generated__/core/DocumentGraphSendSigningNotificationsMutation.graphql"; import { useMutationWithToasts } from "../useMutationWithToasts"; @@ -65,28 +64,6 @@ export function useBulkDeleteDocumentsMutation() { }); } -const sendSigningNotificationsMutation = graphql` - mutation DocumentGraphSendSigningNotificationsMutation( - $input: SendSigningNotificationsInput! - ) { - sendSigningNotifications(input: $input) { - success - } - } -`; - -export function useSendSigningNotificationsMutation() { - const { __ } = useTranslate(); - - return useMutationWithToasts( - sendSigningNotificationsMutation, - { - successMessage: __("Signing notifications sent successfully."), - errorMessage: __("Failed to send signing notifications"), - }, - ); -} - const bulkExportDocumentsMutation = graphql` mutation DocumentGraphBulkExportDocumentsMutation( $input: BulkExportDocumentsInput! diff --git a/apps/console/src/pages/organizations/documents/DocumentsPage.tsx b/apps/console/src/pages/organizations/documents/DocumentsPage.tsx index 985530039..2a2a4dd53 100644 --- a/apps/console/src/pages/organizations/documents/DocumentsPage.tsx +++ b/apps/console/src/pages/organizations/documents/DocumentsPage.tsx @@ -16,7 +16,6 @@ import { usePageTitle } from "@probo/hooks"; import { useTranslate } from "@probo/i18n"; import { Button, - IconBell2, IconPlusLarge, PageHeader, TabItem, @@ -30,9 +29,6 @@ import { import { ConnectionHandler, graphql } from "relay-runtime"; import type { DocumentsPageQuery } from "#/__generated__/core/DocumentsPageQuery.graphql"; -import { - useSendSigningNotificationsMutation, -} from "#/hooks/graph/DocumentGraph"; import { useOrganizationId } from "#/hooks/useOrganizationId"; import { CreateDocumentDialog } from "./_components/CreateDocumentDialog"; @@ -66,11 +62,8 @@ export default function DocumentsPage(props: { throw new Error("invalid type for node"); } - const [sendSigningNotifications] = useSendSigningNotificationsMutation(); - usePageTitle(__("Documents")); - const [canSendAnySignatureNotifications, setCanSendAnySignatureNotifications] = useState(false); const [tab, setTab] = useState<"ACTIVE" | "ARCHIVED">("ACTIVE"); const [documentListConnectionId, setDocumentListConnectionId] = useState( ConnectionHandler.getConnectionID( @@ -80,14 +73,6 @@ export default function DocumentsPage(props: { ), ); - const handleSendSigningNotifications = async () => { - await sendSigningNotifications({ - variables: { - input: { organizationId }, - }, - }); - }; - return (
- {canSendAnySignatureNotifications && ( - - )} {organization.canCreateDocument && tab === "ACTIVE" && (
diff --git a/apps/console/src/pages/organizations/documents/_components/CreateDocumentDialog.tsx b/apps/console/src/pages/organizations/documents/_components/CreateDocumentDialog.tsx index a86118c53..c430bcfa2 100644 --- a/apps/console/src/pages/organizations/documents/_components/CreateDocumentDialog.tsx +++ b/apps/console/src/pages/organizations/documents/_components/CreateDocumentDialog.tsx @@ -59,7 +59,6 @@ const createDocumentMutation = graphql` canRequestSignatures: permission(action: "core:document-version:request-signature") canArchive: permission(action: "core:document:archive") canUnarchive: permission(action: "core:document:unarchive") - canSendSigningNotifications: permission(action: "core:document:send-signing-notifications") ...DocumentListItemFragment } } diff --git a/apps/console/src/pages/organizations/documents/_components/DocumentList.tsx b/apps/console/src/pages/organizations/documents/_components/DocumentList.tsx index 777daab44..9739a9f4d 100644 --- a/apps/console/src/pages/organizations/documents/_components/DocumentList.tsx +++ b/apps/console/src/pages/organizations/documents/_components/DocumentList.tsx @@ -72,9 +72,6 @@ const fragment = graphql` ) canArchive: permission(action: "core:document:archive") canUnarchive: permission(action: "core:document:unarchive") - canSendSigningNotifications: permission( - action: "core:document:send-signing-notifications" - ) ...DocumentListItemFragment } } @@ -115,10 +112,9 @@ const bulkUnarchiveMutation = graphql` export function DocumentList(props: { fKey: DocumentListFragment$key; onConnectionIdChange: (connectionId: string) => void; - onCanSendNotificationsChange?: (can: boolean) => void; tab: "ACTIVE" | "ARCHIVED"; }) { - const { fKey, onConnectionIdChange, onCanSendNotificationsChange, tab } = props; + const { fKey, onConnectionIdChange, tab } = props; const organizationId = useOrganizationId(); const { email: defaultEmail } = use(CurrentUser); @@ -170,19 +166,12 @@ export function DocumentList(props: { const canRequestAnySignatures = documents.some(({ canRequestSignatures }) => canRequestSignatures); const canArchiveAny = documents.some(({ canArchive }) => canArchive); const canUnarchiveAny = documents.some(({ canUnarchive }) => canUnarchive); - const canSendAnySignatureNotifications = documents.some( - ({ canSendSigningNotifications }) => canSendSigningNotifications, - ); const hasAnyAction = tab === "ARCHIVED" ? canUnarchiveAny || canDeleteAny : canArchiveAny || canDeleteAny || canUpdateAny; useEffect(() => { onConnectionIdChange(connectionId); }, [connectionId, onConnectionIdChange]); - useEffect(() => { - onCanSendNotificationsChange?.(canSendAnySignatureNotifications); - }, [canSendAnySignatureNotifications, onCanSendNotificationsChange]); - const handleDocumentTypeFilterChange = (value: string) => { const newType = value === "ALL" ? null : (value as DocumentType); clear(); diff --git a/e2e/console/tracker_pattern_test.go b/e2e/console/tracker_pattern_test.go index 0c88ad285..7241b7d88 100644 --- a/e2e/console/tracker_pattern_test.go +++ b/e2e/console/tracker_pattern_test.go @@ -679,6 +679,7 @@ func seedCommonTrackerPattern(t *testing.T) gid.GID { defer cancel() cleanupConn := dialTestPg(t, cleanupCtx) + defer func() { _ = cleanupConn.Close(cleanupCtx) }() _, err := cleanupConn.Exec(cleanupCtx, `DELETE FROM common_tracker_patterns WHERE id = $1`, id) diff --git a/packages/emails/emails.go b/packages/emails/emails.go index a493c492c..be0b23a9a 100644 --- a/packages/emails/emails.go +++ b/packages/emails/emails.go @@ -20,7 +20,6 @@ import ( "embed" "fmt" htmltemplate "html/template" - "net/url" texttemplate "text/template" "time" @@ -60,6 +59,12 @@ type ( config PresenterConfig RecipientFullName string } + + DocumentSummary struct { + Title string + Type string + URL string + } ) func DefaultPresenterConfig(baseURL string) PresenterConfig { @@ -92,7 +97,7 @@ const ( subjectConfirmEmail = "Confirm your email address" subjectPasswordReset = "Reset your password" subjectInvitation = "Invitation to join %s on Probo" - subjectDocumentApproval = "Action Required – Please review and approve %s" + subjectDocumentApproval = "Action Required – Please review and approve %s compliance documents" subjectDocumentSigning = "Action Required – Please review and sign %s compliance documents" subjectDocumentExport = "Your document export is ready" subjectFrameworkExport = "Your framework export is ready" @@ -231,62 +236,53 @@ func (p *Presenter) RenderInvitation(ctx context.Context, invitationURLPath stri func (p *Presenter) RenderDocumentApproval( ctx context.Context, - approvalURLPath string, - approvalURLQuery url.Values, + approvalURL string, organizationName string, - documentName string, + documents []DocumentSummary, ) (subject string, textBody string, htmlBody *string, err error) { vars, err := p.getCommonVariables() if err != nil { return "", "", nil, fmt.Errorf("cannot get common variables: %w", err) } - approvalURL := baseurl.MustParse(vars.BaseURL). - AppendPath(approvalURLPath). - WithQueryValues(approvalURLQuery). - MustString() - data := struct { *CommonVariables ApprovalUrl string OrganizationName string - DocumentName string + Documents []DocumentSummary }{ CommonVariables: vars, ApprovalUrl: approvalURL, OrganizationName: organizationName, - DocumentName: documentName, + Documents: documents, } textBody, htmlBody, err = renderEmail(documentApprovalTextTemplate, documentApprovalHTMLTemplate, data) - return fmt.Sprintf(subjectDocumentApproval, documentName), textBody, htmlBody, err + return fmt.Sprintf(subjectDocumentApproval, organizationName), textBody, htmlBody, err } func (p *Presenter) RenderDocumentSigning( ctx context.Context, - signingURLPath string, - signingURLQuery url.Values, + signingURL string, organizationName string, + documents []DocumentSummary, ) (subject string, textBody string, htmlBody *string, err error) { vars, err := p.getCommonVariables() if err != nil { return "", "", nil, fmt.Errorf("cannot get common variables: %w", err) } - signingURL := baseurl.MustParse(vars.BaseURL). - AppendPath(signingURLPath). - WithQueryValues(signingURLQuery). - MustString() - data := struct { *CommonVariables SigningUrl string OrganizationName string + Documents []DocumentSummary }{ CommonVariables: vars, SigningUrl: signingURL, OrganizationName: organizationName, + Documents: documents, } textBody, htmlBody, err = renderEmail(documentSigningTextTemplate, documentSigningHTMLTemplate, data) diff --git a/packages/emails/src/DocumentApproval.tsx b/packages/emails/src/DocumentApproval.tsx index 7e1677d80..cabaf6d87 100644 --- a/packages/emails/src/DocumentApproval.tsx +++ b/packages/emails/src/DocumentApproval.tsx @@ -16,14 +16,22 @@ import { Button, Section, Text } from 'react-email'; import * as React from 'react'; import EmailLayout, { bodyText, button, buttonContainer, footerText } from './components/EmailLayout'; +const documentList = { + ...bodyText, + paddingLeft: '20px', +}; + export const DocumentApproval = () => { return ( - + - You're receiving this message because {'{{.OrganizationName}}'} has requested your approval on the document {'{{.DocumentName}}'}. + You're receiving this message because {'{{.OrganizationName}}'} has requested your approval on the following documents: + +
    {{.Title}} ({{.Type}}){{end}}' }} /> + - Please take a moment to review the document and approve or reject it by clicking the button below: + Please take a moment to review them and approve or reject each one by clicking the button below:
    diff --git a/packages/emails/src/DocumentSigning.tsx b/packages/emails/src/DocumentSigning.tsx index 0d9a66ebc..384ff163d 100644 --- a/packages/emails/src/DocumentSigning.tsx +++ b/packages/emails/src/DocumentSigning.tsx @@ -16,14 +16,22 @@ import { Button, Section, Text } from 'react-email'; import * as React from 'react'; import EmailLayout, { bodyText, button, buttonContainer, footerText } from './components/EmailLayout'; +const documentList = { + ...bodyText, + paddingLeft: '20px', +}; + export const DocumentSigning = () => { return ( - You're receiving this message because your company, {'{{.OrganizationName}}'}, has shared a new compliance document that requires your review and signature. + You're receiving this message because your company, {'{{.OrganizationName}}'}, has shared compliance documents that require your review and signature: + +
      {{.Title}} ({{.Type}}){{end}}' }} /> + - To stay compliant with company policies, please take a moment to review and sign the document by clicking the button below: + To stay compliant with company policies, please take a moment to review and sign them by clicking the button below:
      diff --git a/packages/emails/templates/document-approval.txt b/packages/emails/templates/document-approval.txt index fdb03a53c..a25817124 100644 --- a/packages/emails/templates/document-approval.txt +++ b/packages/emails/templates/document-approval.txt @@ -2,9 +2,11 @@ Probo Hi {{.RecipientFullName}}, -You're receiving this message because {{.OrganizationName}} has requested your approval on the document {{.DocumentName}}. +You're receiving this message because {{.OrganizationName}} has requested your approval on the following documents: -Please take a moment to review the document and approve or reject it by clicking the link below: +{{range .Documents}}- {{.Title}} ({{.Type}}): {{.URL}} +{{end}} +Please take a moment to review them and approve or reject each one by clicking the link below: {{.ApprovalUrl}} diff --git a/packages/emails/templates/document-signing.txt b/packages/emails/templates/document-signing.txt index ffbf89811..38b67de3a 100644 --- a/packages/emails/templates/document-signing.txt +++ b/packages/emails/templates/document-signing.txt @@ -2,9 +2,11 @@ Probo Hi {{.RecipientFullName}}, -You're receiving this message because your company, {{.OrganizationName}}, has shared a new compliance document that requires your review and signature. +You're receiving this message because your company, {{.OrganizationName}}, has shared compliance documents that require your review and signature: -To stay compliant with company policies, please take a moment to review and sign the document by clicking the link below: +{{range .Documents}}- {{.Title}} ({{.Type}}): {{.URL}} +{{end}} +To stay compliant with company policies, please take a moment to review and sign them by clicking the link below: {{.SigningUrl}} diff --git a/packages/n8n-node/nodes/Probo/actions/document/index.ts b/packages/n8n-node/nodes/Probo/actions/document/index.ts index 897fec487..8d9201c35 100644 --- a/packages/n8n-node/nodes/Probo/actions/document/index.ts +++ b/packages/n8n-node/nodes/Probo/actions/document/index.ts @@ -31,7 +31,6 @@ import * as getSignatureOp from './getSignature.operation'; import * as getAllSignaturesOp from './getAllSignatures.operation'; import * as requestSignatureOp from './requestSignature.operation'; import * as cancelSignatureOp from './cancelSignature.operation'; -import * as sendSigningNotificationsOp from './sendSigningNotifications.operation'; import * as getApprovalQuorumOp from './getApprovalQuorum.operation'; import * as getAllApprovalQuorumsOp from './getAllApprovalQuorums.operation'; import * as getApprovalDecisionOp from './getApprovalDecision.operation'; @@ -157,12 +156,6 @@ export const description: INodeProperties[] = [ description: 'Request a signature for a document version', action: 'Request a document version signature', }, - { - name: 'Send Signing Notifications', - value: 'sendSigningNotifications', - description: 'Send signing notifications to all pending signatories', - action: 'Send signing notifications', - }, { name: 'Unarchive', value: 'unarchive', @@ -208,7 +201,6 @@ export const description: INodeProperties[] = [ ...getAllSignaturesOp.description, ...requestSignatureOp.description, ...cancelSignatureOp.description, - ...sendSigningNotificationsOp.description, ...getApprovalQuorumOp.description, ...getAllApprovalQuorumsOp.description, ...getApprovalDecisionOp.description, @@ -234,7 +226,6 @@ export { getAllSignaturesOp as getAllSignatures, requestSignatureOp as requestSignature, cancelSignatureOp as cancelSignature, - sendSigningNotificationsOp as sendSigningNotifications, getApprovalQuorumOp as getApprovalQuorum, getAllApprovalQuorumsOp as getAllApprovalQuorums, getApprovalDecisionOp as getApprovalDecision, diff --git a/packages/n8n-node/nodes/Probo/actions/document/sendSigningNotifications.operation.ts b/packages/n8n-node/nodes/Probo/actions/document/sendSigningNotifications.operation.ts deleted file mode 100644 index 4afbbce6c..000000000 --- a/packages/n8n-node/nodes/Probo/actions/document/sendSigningNotifications.operation.ts +++ /dev/null @@ -1,55 +0,0 @@ -// Copyright (c) 2025-2026 Probo Inc . -// -// Permission to use, copy, modify, and/or distribute this software for any -// purpose with or without fee is hereby granted, provided that the above -// copyright notice and this permission notice appear in all copies. -// -// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH -// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY -// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, -// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM -// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR -// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR -// PERFORMANCE OF THIS SOFTWARE. - -import type { INodeProperties, IExecuteFunctions, INodeExecutionData } from 'n8n-workflow'; -import { proboApiRequest } from '../../GenericFunctions'; - -export const description: INodeProperties[] = [ - { - displayName: 'Organization ID', - name: 'organizationId', - type: 'string', - displayOptions: { - show: { - resource: ['document'], - operation: ['sendSigningNotifications'], - }, - }, - default: '', - description: 'The ID of the organization', - required: true, - }, -]; - -export async function execute( - this: IExecuteFunctions, - itemIndex: number, -): Promise { - const organizationId = this.getNodeParameter('organizationId', itemIndex) as string; - - const query = ` - mutation SendSigningNotifications($input: SendSigningNotificationsInput!) { - sendSigningNotifications(input: $input) { - success - } - } - `; - - const responseData = await proboApiRequest.call(this, query, { input: { organizationId } }); - - return { - json: responseData, - pairedItem: { item: itemIndex }, - }; -} diff --git a/pkg/bootstrap/builder.go b/pkg/bootstrap/builder.go index a734b1d26..f7d16e42c 100644 --- a/pkg/bootstrap/builder.go +++ b/pkg/bootstrap/builder.go @@ -177,6 +177,11 @@ func (b *Builder) Build() (*probodconfig.FullConfig, error) { SenderInterval: b.getEnvIntOrDefault("WEBHOOK_SENDER_INTERVAL", 5), CacheTTL: b.getEnvIntOrDefault("WEBHOOK_CACHE_TTL", 86400), }, + Document: probodconfig.DocumentNotificationConfig{ + Interval: b.getEnvIntOrDefault("DOCUMENT_NOTIFICATION_INTERVAL", 300), + DebounceDelay: b.getEnvIntOrDefault("DOCUMENT_NOTIFICATION_DEBOUNCE_DELAY", 900), + ReminderInterval: b.getEnvIntOrDefault("DOCUMENT_NOTIFICATION_REMINDER_INTERVAL", 86400), + }, }, Agents: probodconfig.AgentsConfig{ Providers: map[string]probodconfig.LLMProviderConfig{ diff --git a/pkg/bootstrap/builder_test.go b/pkg/bootstrap/builder_test.go index 78496f407..4019d6e7b 100644 --- a/pkg/bootstrap/builder_test.go +++ b/pkg/bootstrap/builder_test.go @@ -198,6 +198,9 @@ func TestBuilder_Build_Defaults(t *testing.T) { assert.Empty(t, cfg.Probod.Notifications.Slack.SigningSecret) assert.Equal(t, 5, cfg.Probod.Notifications.Webhook.SenderInterval) assert.Equal(t, 86400, cfg.Probod.Notifications.Webhook.CacheTTL) + assert.Equal(t, 300, cfg.Probod.Notifications.Document.Interval) + assert.Equal(t, 900, cfg.Probod.Notifications.Document.DebounceDelay) + assert.Equal(t, 86400, cfg.Probod.Notifications.Document.ReminderInterval) // Agents tools — Firecrawl empty by default assert.Empty(t, cfg.Probod.Agents.Tools.FirecrawlAPIKey) @@ -329,6 +332,9 @@ func TestBuilder_Build_CustomValues(t *testing.T) { env["WEBHOOK_SENDER_INTERVAL"] = "10" env["WEBHOOK_CACHE_TTL"] = "3600" env["CONNECTOR_SLACK_SIGNING_SECRET"] = "slack-signing-secret" + env["DOCUMENT_NOTIFICATION_INTERVAL"] = "120" + env["DOCUMENT_NOTIFICATION_DEBOUNCE_DELAY"] = "60" + env["DOCUMENT_NOTIFICATION_REMINDER_INTERVAL"] = "43200" // Firecrawl env["FIRECRAWL_API_KEY"] = "fc-test-key" // Agents — providers @@ -456,6 +462,9 @@ func TestBuilder_Build_CustomValues(t *testing.T) { assert.Equal(t, "slack-signing-secret", cfg.Probod.Notifications.Slack.SigningSecret) assert.Equal(t, 10, cfg.Probod.Notifications.Webhook.SenderInterval) assert.Equal(t, 3600, cfg.Probod.Notifications.Webhook.CacheTTL) + assert.Equal(t, 120, cfg.Probod.Notifications.Document.Interval) + assert.Equal(t, 60, cfg.Probod.Notifications.Document.DebounceDelay) + assert.Equal(t, 43200, cfg.Probod.Notifications.Document.ReminderInterval) // Agents tools — Firecrawl assert.Equal(t, "fc-test-key", cfg.Probod.Agents.Tools.FirecrawlAPIKey) // Agents — providers diff --git a/pkg/coredata/document_type.go b/pkg/coredata/document_type.go index 17217948b..ae6e9ac9b 100644 --- a/pkg/coredata/document_type.go +++ b/pkg/coredata/document_type.go @@ -80,6 +80,31 @@ func (v DocumentType) String() string { return string(v) } +func (v DocumentType) Label() string { + switch v { + case DocumentTypeGovernance: + return "Governance" + case DocumentTypePolicy: + return "Policy" + case DocumentTypeProcedure: + return "Procedure" + case DocumentTypePlan: + return "Plan" + case DocumentTypeRegister: + return "Register" + case DocumentTypeRecord: + return "Record" + case DocumentTypeReport: + return "Report" + case DocumentTypeTemplate: + return "Template" + case DocumentTypeStatementOfApplicability: + return "Statement of Applicability" + default: + return "Document" + } +} + func (v DocumentType) MarshalText() ([]byte, error) { return []byte(v.String()), nil } diff --git a/pkg/coredata/document_version.go b/pkg/coredata/document_version.go index 4fe3c273c..6247c24bc 100644 --- a/pkg/coredata/document_version.go +++ b/pkg/coredata/document_version.go @@ -152,6 +152,58 @@ WHERE return nil } +func (dv *DocumentVersions) LoadByIDs( + ctx context.Context, + conn pg.Querier, + scope Scoper, + documentVersionIDs []gid.GID, +) error { + q := ` +SELECT + id, + organization_id, + document_id, + title, + major, + minor, + classification, + document_type, + content, + changelog, + status, + orientation, + file_id, + pdf_attempt_count, + published_at, + created_at, + updated_at +FROM + document_versions +WHERE + %s + AND id = ANY(@document_version_ids) +` + + q = fmt.Sprintf(q, scope.SQLFragment()) + + args := pgx.StrictNamedArgs{"document_version_ids": documentVersionIDs} + maps.Copy(args, scope.SQLArguments()) + + rows, err := conn.Query(ctx, q, args) + if err != nil { + return fmt.Errorf("cannot query document versions: %w", err) + } + + documentVersions, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[DocumentVersion]) + if err != nil { + return fmt.Errorf("cannot collect document versions: %w", err) + } + + *dv = documentVersions + + return nil +} + func (dv DocumentVersion) CursorKey(orderBy DocumentVersionOrderField) page.CursorKey { switch orderBy { case DocumentVersionOrderFieldCreatedAt: diff --git a/pkg/coredata/document_version_approval_decision.go b/pkg/coredata/document_version_approval_decision.go index 2c413f8eb..1ea86efca 100644 --- a/pkg/coredata/document_version_approval_decision.go +++ b/pkg/coredata/document_version_approval_decision.go @@ -298,7 +298,8 @@ INSERT INTO document_version_approval_decisions ( electronic_signature_id, decided_at, created_at, - updated_at + updated_at, + notification_count ) VALUES ( @id, @tenant_id, @@ -310,7 +311,8 @@ INSERT INTO document_version_approval_decisions ( @electronic_signature_id, @decided_at, @created_at, - @updated_at + @updated_at, + @notification_count ) ` @@ -326,6 +328,7 @@ INSERT INTO document_version_approval_decisions ( "decided_at": d.DecidedAt, "created_at": d.CreatedAt, "updated_at": d.UpdatedAt, + "notification_count": 0, } _, err := conn.Exec(ctx, q, args) @@ -367,6 +370,7 @@ func (ds DocumentVersionApprovalDecisions) BulkInsert( d.DecidedAt, d.CreatedAt, d.UpdatedAt, + 0, }, ) } @@ -386,6 +390,7 @@ func (ds DocumentVersionApprovalDecisions) BulkInsert( "decided_at", "created_at", "updated_at", + "notification_count", }, pgx.CopyFromRows(rows), ) @@ -393,6 +398,211 @@ func (ds DocumentVersionApprovalDecisions) BulkInsert( return err } +// LoadNextDueGroupForNotification loads every still-PENDING approval decision +// for the next (organization, approver) group that has at least one decision due +// for a notification, so the whole group can be emailed together. A decision is +// due once it is past its scheduled offset (see documentNotificationMaxCount) +// and has not reached the cap. The receiver is left empty when no group is due. +// +// A decision is only ever PENDING while its quorum is pending — resolving a +// quorum either approves all decisions or voids the remaining ones — so there is +// no need to join the quorum table here. +func (d *DocumentVersionApprovalDecisions) LoadNextDueGroupForNotification( + ctx context.Context, + conn pg.Querier, + now time.Time, + debounceBefore time.Time, + reminderInterval time.Duration, +) error { + q := ` +WITH next_group AS ( + SELECT + organization_id, + approver_id + FROM + document_version_approval_decisions + WHERE + state = @state + AND notification_count < @max_notifications + AND ( + (notification_count = 0 AND created_at < @debounce_before) + OR (notification_count > 0 AND last_notified_at < @now::timestamptz - make_interval(secs => @reminder_interval_seconds * notification_count)) + ) + AND EXISTS ( + SELECT 1 + FROM document_version_approval_quorums q + JOIN document_versions dv ON dv.id = q.version_id + JOIN documents doc ON doc.id = dv.document_id + WHERE q.id = document_version_approval_decisions.quorum_id + AND doc.deleted_at IS NULL + AND doc.archived_at IS NULL + ) + GROUP BY + organization_id, + approver_id + ORDER BY + organization_id, + approver_id + LIMIT 1 +) +SELECT + d.id, + d.organization_id, + d.quorum_id, + d.approver_id, + d.state, + d.comment, + d.electronic_signature_id, + d.decided_at, + d.created_at, + d.updated_at +FROM + document_version_approval_decisions d +INNER JOIN next_group g + ON g.organization_id = d.organization_id + AND g.approver_id = d.approver_id +WHERE + d.state = @state + AND d.notification_count < @max_notifications + AND EXISTS ( + SELECT 1 + FROM document_version_approval_quorums q + JOIN document_versions dv ON dv.id = q.version_id + JOIN documents doc ON doc.id = dv.document_id + WHERE q.id = d.quorum_id + AND doc.deleted_at IS NULL + AND doc.archived_at IS NULL + ) +ORDER BY + d.quorum_id +` + + args := pgx.StrictNamedArgs{ + "state": DocumentVersionApprovalDecisionStatePending, + "max_notifications": documentNotificationMaxCount, + "now": now, + "debounce_before": debounceBefore, + "reminder_interval_seconds": reminderInterval.Seconds(), + } + + rows, err := conn.Query(ctx, q, args) + if err != nil { + return fmt.Errorf("cannot query due approval decisions for notification: %w", err) + } + + decisions, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[DocumentVersionApprovalDecision]) + if err != nil { + return fmt.Errorf("cannot collect due approval decisions for notification: %w", err) + } + + *d = decisions + + return nil +} + +// ClaimForNotification claims the receiver's decisions that are individually +// due for a notification and returns their ids. The conditional update doubles +// as the claim, so concurrent workers never email the same group twice. Callers +// advance the rest of the group with BumpRemainingForNotification in the same +// transaction. +func (d DocumentVersionApprovalDecisions) ClaimForNotification( + ctx context.Context, + conn pg.Tx, + now time.Time, + debounceBefore time.Time, + reminderInterval time.Duration, +) ([]gid.GID, error) { + ids := make([]gid.GID, len(d)) + for i, decision := range d { + ids[i] = decision.ID + } + + q := ` +UPDATE document_version_approval_decisions +SET + notification_count = notification_count + 1, + last_notified_at = @now +WHERE + id = ANY(@ids::text[]) + AND state = @state + AND notification_count < @max_notifications + AND ( + (notification_count = 0 AND created_at < @debounce_before) + OR (notification_count > 0 AND last_notified_at < @now::timestamptz - make_interval(secs => @reminder_interval_seconds * notification_count)) + ) +RETURNING id +` + + rows, err := conn.Query(ctx, q, pgx.StrictNamedArgs{ + "ids": ids, + "state": DocumentVersionApprovalDecisionStatePending, + "max_notifications": documentNotificationMaxCount, + "now": now, + "debounce_before": debounceBefore, + "reminder_interval_seconds": reminderInterval.Seconds(), + }) + if err != nil { + return nil, fmt.Errorf("cannot claim due approval decisions for notification: %w", err) + } + + claimed, err := pgx.CollectRows(rows, pgx.RowTo[gid.GID]) + if err != nil { + return nil, fmt.Errorf("cannot collect claimed approval decisions for notification: %w", err) + } + + return claimed, nil +} + +// BumpRemainingForNotification advances the notification schedule for the +// still-pending decisions in the group that were not individually claimed, so +// the whole emailed list moves forward together. It must run in the same +// transaction as ClaimForNotification. +func (d DocumentVersionApprovalDecisions) BumpRemainingForNotification( + ctx context.Context, + conn pg.Tx, + claimed []gid.GID, + now time.Time, +) ([]gid.GID, error) { + ids := make([]gid.GID, len(d)) + for i, decision := range d { + ids[i] = decision.ID + } + + rest := remainingNotificationIDs(ids, claimed) + if len(rest) == 0 { + return nil, nil + } + + q := ` +UPDATE document_version_approval_decisions +SET + notification_count = notification_count + 1, + last_notified_at = @now +WHERE + id = ANY(@ids::text[]) + AND state = @state + AND notification_count < @max_notifications +RETURNING id +` + + rows, err := conn.Query(ctx, q, pgx.StrictNamedArgs{ + "ids": rest, + "state": DocumentVersionApprovalDecisionStatePending, + "max_notifications": documentNotificationMaxCount, + "now": now, + }) + if err != nil { + return nil, fmt.Errorf("cannot bump remaining approval decisions for notification: %w", err) + } + + bumped, err := pgx.CollectRows(rows, pgx.RowTo[gid.GID]) + if err != nil { + return nil, fmt.Errorf("cannot collect bumped approval decisions for notification: %w", err) + } + + return bumped, nil +} + func (d *DocumentVersionApprovalDecision) Update( ctx context.Context, conn pg.Tx, diff --git a/pkg/coredata/document_version_approval_quorum.go b/pkg/coredata/document_version_approval_quorum.go index 8bcc58bcc..1bc3dd230 100644 --- a/pkg/coredata/document_version_approval_quorum.go +++ b/pkg/coredata/document_version_approval_quorum.go @@ -135,6 +135,47 @@ WHERE return nil } +func (q *DocumentVersionApprovalQuorums) LoadByIDs( + ctx context.Context, + conn pg.Querier, + scope Scoper, + quorumIDs []gid.GID, +) error { + query := ` +SELECT + id, + organization_id, + version_id, + status, + created_at, + updated_at +FROM + document_version_approval_quorums +WHERE + %s + AND id = ANY(@quorum_ids) +` + + query = fmt.Sprintf(query, scope.SQLFragment()) + + args := pgx.StrictNamedArgs{"quorum_ids": quorumIDs} + maps.Copy(args, scope.SQLArguments()) + + rows, err := conn.Query(ctx, query, args) + if err != nil { + return fmt.Errorf("cannot query approval quorums: %w", err) + } + + quorums, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[DocumentVersionApprovalQuorum]) + if err != nil { + return fmt.Errorf("cannot collect approval quorums: %w", err) + } + + *q = quorums + + return nil +} + func (q *DocumentVersionApprovalQuorum) LoadLastByDocumentVersionID( ctx context.Context, conn pg.Querier, diff --git a/pkg/coredata/document_version_signature.go b/pkg/coredata/document_version_signature.go index 85ffe0d97..c34fbe3bc 100644 --- a/pkg/coredata/document_version_signature.go +++ b/pkg/coredata/document_version_signature.go @@ -297,7 +297,8 @@ INSERT INTO document_version_signatures ( requested_at, electronic_signature_id, created_at, - updated_at + updated_at, + notification_count ) VALUES ( @id, @tenant_id, @@ -309,7 +310,8 @@ INSERT INTO document_version_signatures ( @requested_at, @electronic_signature_id, @created_at, - @updated_at + @updated_at, + @notification_count ) ` @@ -325,6 +327,7 @@ INSERT INTO document_version_signatures ( "electronic_signature_id": pvs.ElectronicSignatureID, "created_at": pvs.CreatedAt, "updated_at": pvs.UpdatedAt, + "notification_count": 0, } _, err := conn.Exec(ctx, q, args) @@ -399,6 +402,230 @@ WHERE return nil } +// documentNotificationMaxCount caps how many emails a signature/approval +// request gets: the first notice plus three reminders. A request is due for its +// next email once it is past its scheduled offset — the first email after the +// debounce delay, then reminders at 1x, 2x and 3x the reminder interval after +// the previous email — and stops once it reaches this cap. +const documentNotificationMaxCount = 4 + +func remainingNotificationIDs(all []gid.GID, claimed []gid.GID) []gid.GID { + claimedSet := make(map[gid.GID]struct{}, len(claimed)) + for _, id := range claimed { + claimedSet[id] = struct{}{} + } + + rest := make([]gid.GID, 0, len(all)) + for _, id := range all { + if _, ok := claimedSet[id]; ok { + continue + } + + rest = append(rest, id) + } + + return rest +} + +// LoadNextDueGroupForNotification loads every still-REQUESTED signature for the +// next (organization, signatory) group that has at least one request due for a +// notification, so the whole group can be emailed together. A request is due +// once it is past its scheduled offset (see documentNotificationMaxCount) and +// has not reached the cap. The receiver is left empty when no group is due. +func (pvss *DocumentVersionSignatures) LoadNextDueGroupForNotification( + ctx context.Context, + conn pg.Querier, + now time.Time, + debounceBefore time.Time, + reminderInterval time.Duration, +) error { + q := ` +WITH next_group AS ( + SELECT + organization_id, + signed_by_profile_id + FROM + document_version_signatures + WHERE + state = @state + AND notification_count < @max_notifications + AND ( + (notification_count = 0 AND requested_at < @debounce_before) + OR (notification_count > 0 AND last_notified_at < @now::timestamptz - make_interval(secs => @reminder_interval_seconds * notification_count)) + ) + AND EXISTS ( + SELECT 1 + FROM document_versions dv + JOIN documents doc ON doc.id = dv.document_id + WHERE dv.id = document_version_signatures.document_version_id + AND doc.deleted_at IS NULL + AND doc.archived_at IS NULL + ) + GROUP BY + organization_id, + signed_by_profile_id + ORDER BY + organization_id, + signed_by_profile_id + LIMIT 1 +) +SELECT + s.id, + s.organization_id, + s.document_version_id, + s.state, + s.signed_by_profile_id, + s.signed_at, + s.requested_at, + s.electronic_signature_id, + s.created_at, + s.updated_at +FROM + document_version_signatures s +INNER JOIN next_group g + ON g.organization_id = s.organization_id + AND g.signed_by_profile_id = s.signed_by_profile_id +WHERE + s.state = @state + AND s.notification_count < @max_notifications + AND EXISTS ( + SELECT 1 + FROM document_versions dv + JOIN documents doc ON doc.id = dv.document_id + WHERE dv.id = s.document_version_id + AND doc.deleted_at IS NULL + AND doc.archived_at IS NULL + ) +ORDER BY + s.document_version_id +` + + args := pgx.StrictNamedArgs{ + "state": DocumentVersionSignatureStateRequested, + "max_notifications": documentNotificationMaxCount, + "now": now, + "debounce_before": debounceBefore, + "reminder_interval_seconds": reminderInterval.Seconds(), + } + + rows, err := conn.Query(ctx, q, args) + if err != nil { + return fmt.Errorf("cannot query due signatures for notification: %w", err) + } + + signatures, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[DocumentVersionSignature]) + if err != nil { + return fmt.Errorf("cannot collect due signatures for notification: %w", err) + } + + *pvss = signatures + + return nil +} + +// ClaimForNotification claims the receiver's signatures that are individually +// due for a notification and returns their ids. The conditional update doubles +// as the claim, so concurrent workers never email the same group twice. Callers +// advance the rest of the group with BumpRemainingForNotification in the same +// transaction. +func (pvss DocumentVersionSignatures) ClaimForNotification( + ctx context.Context, + conn pg.Tx, + now time.Time, + debounceBefore time.Time, + reminderInterval time.Duration, +) ([]gid.GID, error) { + ids := make([]gid.GID, len(pvss)) + for i, signature := range pvss { + ids[i] = signature.ID + } + + q := ` +UPDATE document_version_signatures +SET + notification_count = notification_count + 1, + last_notified_at = @now +WHERE + id = ANY(@ids::text[]) + AND state = @state + AND notification_count < @max_notifications + AND ( + (notification_count = 0 AND requested_at < @debounce_before) + OR (notification_count > 0 AND last_notified_at < @now::timestamptz - make_interval(secs => @reminder_interval_seconds * notification_count)) + ) +RETURNING id +` + + rows, err := conn.Query(ctx, q, pgx.StrictNamedArgs{ + "ids": ids, + "state": DocumentVersionSignatureStateRequested, + "max_notifications": documentNotificationMaxCount, + "now": now, + "debounce_before": debounceBefore, + "reminder_interval_seconds": reminderInterval.Seconds(), + }) + if err != nil { + return nil, fmt.Errorf("cannot claim due signatures for notification: %w", err) + } + + claimed, err := pgx.CollectRows(rows, pgx.RowTo[gid.GID]) + if err != nil { + return nil, fmt.Errorf("cannot collect claimed signatures for notification: %w", err) + } + + return claimed, nil +} + +// BumpRemainingForNotification advances the notification schedule for the +// still-pending signatures in the group that were not individually claimed, so +// the whole emailed list moves forward together. It must run in the same +// transaction as ClaimForNotification. +func (pvss DocumentVersionSignatures) BumpRemainingForNotification( + ctx context.Context, + conn pg.Tx, + claimed []gid.GID, + now time.Time, +) ([]gid.GID, error) { + ids := make([]gid.GID, len(pvss)) + for i, signature := range pvss { + ids[i] = signature.ID + } + + rest := remainingNotificationIDs(ids, claimed) + if len(rest) == 0 { + return nil, nil + } + + q := ` +UPDATE document_version_signatures +SET + notification_count = notification_count + 1, + last_notified_at = @now +WHERE + id = ANY(@ids::text[]) + AND state = @state + AND notification_count < @max_notifications +RETURNING id +` + + rows, err := conn.Query(ctx, q, pgx.StrictNamedArgs{ + "ids": rest, + "state": DocumentVersionSignatureStateRequested, + "max_notifications": documentNotificationMaxCount, + "now": now, + }) + if err != nil { + return nil, fmt.Errorf("cannot bump remaining signatures for notification: %w", err) + } + + bumped, err := pgx.CollectRows(rows, pgx.RowTo[gid.GID]) + if err != nil { + return nil, fmt.Errorf("cannot collect bumped signatures for notification: %w", err) + } + + return bumped, nil +} + func (pvs *DocumentVersionSignature) Update( ctx context.Context, conn pg.Tx, diff --git a/pkg/coredata/membership_profile.go b/pkg/coredata/membership_profile.go index 888a021f4..c35777f4d 100644 --- a/pkg/coredata/membership_profile.go +++ b/pkg/coredata/membership_profile.go @@ -838,82 +838,6 @@ WHERE return count, nil } -func (p *MembershipProfiles) LoadAwaitingSigning( - ctx context.Context, - conn pg.Querier, - scope Scoper, -) error { - q := ` -WITH signatories AS ( - SELECT - signed_by_profile_id - FROM - document_version_signatures - WHERE - %s - AND state = 'REQUESTED' - GROUP BY - signed_by_profile_id -) -SELECT - p.id, - p.identity_id, - p.organization_id, - p.kind, - p.full_name, - i.email_address, - p.source, - p.state, - p.additional_email_addresses, - p.position, - p.contract_start_date, - p.contract_end_date, - '' AS organization_name, - p.user_name, - p.external_id, - p.nickname, - p.locale, - p.timezone, - p.profile_url, - p.preferred_language, - p.given_name, - p.family_name, - p.formatted_name, - p.middle_name, - p.honorific_prefix, - p.honorific_suffix, - p.employee_number, - p.department, - p.cost_center, - p.enterprise_organization, - p.division, - p.manager_value, - p.created_at, - p.updated_at -FROM - iam_membership_profiles p -INNER JOIN identities i - ON i.id = p.identity_id -INNER JOIN signatories ON p.id = signatories.signed_by_profile_id -` - - q = fmt.Sprintf(q, scope.SQLFragment()) - - rows, err := conn.Query(ctx, q, scope.SQLArguments()) - if err != nil { - return fmt.Errorf("cannot query profiles: %w", err) - } - - profiles, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[MembershipProfile]) - if err != nil { - return fmt.Errorf("cannot collect profiles: %w", err) - } - - *p = profiles - - return nil -} - func (p *MembershipProfiles) CountByIdentityID( ctx context.Context, conn pg.Querier, diff --git a/pkg/coredata/migrations/20260612T150000Z.sql b/pkg/coredata/migrations/20260612T150000Z.sql new file mode 100644 index 000000000..f5efb9fcc --- /dev/null +++ b/pkg/coredata/migrations/20260612T150000Z.sql @@ -0,0 +1,31 @@ +-- Copyright (c) 2026 Probo Inc . +-- +-- Permission to use, copy, modify, and/or distribute this software for any +-- purpose with or without fee is hereby granted, provided that the above +-- copyright notice and this permission notice appear in all copies. +-- +-- THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +-- REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +-- AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +-- INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +-- LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +-- OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +-- PERFORMANCE OF THIS SOFTWARE. + +ALTER TABLE document_version_signatures + ADD COLUMN last_notified_at TIMESTAMPTZ; + +ALTER TABLE document_version_signatures + ADD COLUMN notification_count INTEGER NOT NULL DEFAULT 0; + +ALTER TABLE document_version_signatures + ALTER COLUMN notification_count DROP DEFAULT; + +ALTER TABLE document_version_approval_decisions + ADD COLUMN last_notified_at TIMESTAMPTZ; + +ALTER TABLE document_version_approval_decisions + ADD COLUMN notification_count INTEGER NOT NULL DEFAULT 0; + +ALTER TABLE document_version_approval_decisions + ALTER COLUMN notification_count DROP DEFAULT; diff --git a/pkg/probo/actions.go b/pkg/probo/actions.go index 1751c92d1..fb432215b 100644 --- a/pkg/probo/actions.go +++ b/pkg/probo/actions.go @@ -183,16 +183,15 @@ const ( ActionEvidenceDelete = "core:evidence:delete" // Document actions - ActionDocumentGet = "core:document:get" - ActionDocumentList = "core:document:list" - ActionDocumentCreate = "core:document:create" - ActionDocumentUpdate = "core:document:update" - ActionDocumentDelete = "core:document:delete" - ActionDocumentChangelogGenerate = "core:document:generate-changelog" - ActionDocumentArchive = "core:document:archive" - ActionDocumentUnarchive = "core:document:unarchive" - ActionDocumentDeleteDraft = "core:document:delete-draft" - ActionDocumentSendSigningNotifications = "core:document:send-signing-notifications" + ActionDocumentGet = "core:document:get" + ActionDocumentList = "core:document:list" + ActionDocumentCreate = "core:document:create" + ActionDocumentUpdate = "core:document:update" + ActionDocumentDelete = "core:document:delete" + ActionDocumentChangelogGenerate = "core:document:generate-changelog" + ActionDocumentArchive = "core:document:archive" + ActionDocumentUnarchive = "core:document:unarchive" + ActionDocumentDeleteDraft = "core:document:delete-draft" // DocumentVersion actions ActionDocumentVersionGet = "core:document-version:get" diff --git a/pkg/probo/document_approval_service.go b/pkg/probo/document_approval_service.go index ffab27187..694ed5cbd 100644 --- a/pkg/probo/document_approval_service.go +++ b/pkg/probo/document_approval_service.go @@ -19,21 +19,16 @@ import ( "context" "errors" "fmt" - "net/url" "time" "go.gearno.de/crypto/uuid" "go.gearno.de/kit/pg" - "go.probo.inc/probo/packages/emails" - "go.probo.inc/probo/pkg/baseurl" "go.probo.inc/probo/pkg/coredata" "go.probo.inc/probo/pkg/esign" "go.probo.inc/probo/pkg/gid" "go.probo.inc/probo/pkg/html2pdf" - "go.probo.inc/probo/pkg/iam" "go.probo.inc/probo/pkg/mail" "go.probo.inc/probo/pkg/page" - "go.probo.inc/probo/pkg/statelesstoken" ) const DocumentApprovalConsentText = "By clicking \"Review and approve\", I consent to approve this document electronically and agree that my electronic signature has the same legal validity as a handwritten signature." @@ -82,16 +77,6 @@ func (s *DocumentApprovalService) RequestApprovalInTx( approverIDs []gid.GID, changelog *string, ) (*coredata.DocumentVersionApprovalQuorum, error) { - organization := &coredata.Organization{} - if err := organization.LoadByID(ctx, tx, scope, document.OrganizationID); err != nil { - return nil, fmt.Errorf("cannot load organization: %w", err) - } - - approverProfiles := &coredata.MembershipProfiles{} - if err := approverProfiles.LoadByIDs(ctx, tx, scope, approverIDs); err != nil { - return nil, fmt.Errorf("cannot load approver profiles: %w", err) - } - now := time.Now() documentVersion.Status = coredata.DocumentVersionStatusPendingApproval @@ -129,9 +114,9 @@ func (s *DocumentApprovalService) RequestApprovalInTx( return nil, fmt.Errorf("cannot create approval decisions: %w", err) } - if err := s.sendApprovalEmails(ctx, scope, tx, *approverProfiles, document, organization, documentVersion.ID); err != nil { - return nil, fmt.Errorf("cannot send approval emails: %w", err) - } + // Approval notifications are sent asynchronously and debounced by the + // document notification worker, which batches all pending approvals per + // recipient into a single email. return quorum, nil } @@ -809,88 +794,6 @@ func (s *DocumentApprovalService) createDecisions( return nil } -func (s *DocumentApprovalService) sendApprovalEmails( - ctx context.Context, scope coredata.Scoper, - tx pg.Tx, - profiles coredata.MembershipProfiles, - document *coredata.Document, - organization *coredata.Organization, - documentVersionID gid.GID, -) error { - now := time.Now() - approvalURLPath := "/organizations/" + document.OrganizationID.String() + "/employee/approvals/" + document.ID.String() - - approvalEmails := make(coredata.Emails, 0, len(profiles)) - for _, profile := range profiles { - emailPresenter := emails.NewPresenter(s.svc.baseURL, profile.FullName) - - var ( - emailLinkURLPath = approvalURLPath - query = make(url.Values) - ) - - if profile.State != coredata.ProfileStateActive { - if profile.Source != coredata.ProfileSourceSCIM { - invitation := &coredata.Invitation{ - ID: gid.New(document.OrganizationID.TenantID(), coredata.InvitationEntityType), - OrganizationID: document.OrganizationID, - UserID: profile.ID, - Status: coredata.InvitationStatusPending, - ExpiresAt: now.Add(s.invitationTokenValidity), - CreatedAt: now, - } - if err := invitation.Insert(ctx, tx, coredata.NewScopeFromObjectID(document.OrganizationID)); err != nil { - return fmt.Errorf("cannot insert invitation: %w", err) - } - - invitationToken, err := statelesstoken.NewToken( - s.tokenSecret, - iam.TokenTypeOrganizationInvitation, - s.invitationTokenValidity, - iam.InvitationTokenData{InvitationID: invitation.ID}, - ) - if err != nil { - return fmt.Errorf("cannot generate invitation token: %w", err) - } - - emailLinkURLPath = "/auth/activate-account" - continueURL := baseurl.MustParse(s.svc.baseURL).AppendPath(approvalURLPath).MustString() - - query.Add("token", invitationToken) - query.Add("continue", continueURL) - } - } - - subject, textBody, htmlBody, err := emailPresenter.RenderDocumentApproval( - ctx, - emailLinkURLPath, - query, - organization.Name, - document.Title, - ) - if err != nil { - return fmt.Errorf("cannot render approval request email: %w", err) - } - - approvalEmails = append(approvalEmails, coredata.NewEmail( - profile.FullName, - profile.EmailAddress, - subject, - textBody, - htmlBody, - &coredata.EmailOptions{ - SenderName: new(organization.Name), - }, - )) - } - - if err := approvalEmails.BulkInsert(ctx, tx); err != nil { - return fmt.Errorf("cannot insert approval emails: %w", err) - } - - return nil -} - func (s *DocumentApprovalService) generateApprovalPDF( ctx context.Context, scope coredata.Scoper, documentVersionID gid.GID, diff --git a/pkg/probo/document_notification_worker.go b/pkg/probo/document_notification_worker.go new file mode 100644 index 000000000..33295f926 --- /dev/null +++ b/pkg/probo/document_notification_worker.go @@ -0,0 +1,549 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +package probo + +import ( + "context" + "fmt" + "net/url" + "sort" + "time" + + "go.gearno.de/kit/log" + "go.gearno.de/kit/pg" + "go.gearno.de/kit/worker" + "go.probo.inc/probo/packages/emails" + "go.probo.inc/probo/pkg/baseurl" + "go.probo.inc/probo/pkg/coredata" + "go.probo.inc/probo/pkg/gid" + "go.probo.inc/probo/pkg/iam" + "go.probo.inc/probo/pkg/statelesstoken" +) + +type ( + notificationKind string + + claimStatus string + + // documentNotificationTask is one consolidated email to send: every document + // awaiting a given recipient's signature or approval in one organization. The + // schedule has already been advanced (the requests are claimed) by the time a + // task is handed to Process. + documentNotificationTask struct { + kind notificationKind + organizationID gid.GID + recipientID gid.GID + versionIDs []gid.GID + } + + DocumentNotificationWorkerConfig struct { + DebounceDelay time.Duration + ReminderInterval time.Duration + } + + documentNotificationHandler struct { + service *Service + logger *log.Logger + debounceDelay time.Duration + reminderInterval time.Duration + } +) + +const ( + notificationKindSigning notificationKind = "signing" + notificationKindApproval notificationKind = "approval" + + claimStatusNone claimStatus = "none" + claimStatusClaimed claimStatus = "claimed" + claimStatusRaced claimStatus = "raced" +) + +// NewDocumentNotificationWorker builds the worker that emails recipients, one +// consolidated message per organization, about the documents awaiting their +// signature or approval. Each claim advances the request's reminder schedule, so +// the conditional update doubles as the claim and concurrent workers never email +// the same group twice. +func NewDocumentNotificationWorker( + service *Service, + logger *log.Logger, + cfg DocumentNotificationWorkerConfig, + opts ...worker.Option, +) *worker.Worker[documentNotificationTask] { + h := &documentNotificationHandler{ + service: service, + logger: logger, + debounceDelay: cfg.DebounceDelay, + reminderInterval: cfg.ReminderInterval, + } + + return worker.New( + "document-notification-worker", + h, + logger, + opts..., + ) +} + +func (h *documentNotificationHandler) Claim(ctx context.Context) (documentNotificationTask, error) { + now := time.Now() + debounceBefore := now.Add(-h.debounceDelay) + + task, status, err := h.claimNextSigningGroup(ctx, now, debounceBefore) + if err != nil { + return documentNotificationTask{}, err + } + + for status == claimStatusRaced { + task, status, err = h.claimNextSigningGroup(ctx, now, debounceBefore) + if err != nil { + return documentNotificationTask{}, err + } + } + + if status == claimStatusClaimed { + return task, nil + } + + task, status, err = h.claimNextApprovalGroup(ctx, now, debounceBefore) + if err != nil { + return documentNotificationTask{}, err + } + + for status == claimStatusRaced { + task, status, err = h.claimNextApprovalGroup(ctx, now, debounceBefore) + if err != nil { + return documentNotificationTask{}, err + } + } + + if status == claimStatusClaimed { + return task, nil + } + + return documentNotificationTask{}, worker.ErrNoTask +} + +func (h *documentNotificationHandler) Process(ctx context.Context, task documentNotificationTask) error { + if len(task.versionIDs) == 0 { + return nil + } + + scope := coredata.NewScopeFromObjectID(task.organizationID) + + if err := h.service.pg.WithTx( + ctx, + func(ctx context.Context, tx pg.Tx) error { + return h.service.Documents.sendNotification(ctx, tx, scope, task.kind, task.organizationID, task.recipientID, task.versionIDs) + }, + ); err != nil { + h.logger.ErrorCtx( + ctx, + "document notification worker failure", + log.Error(err), + log.String("organization_id", task.organizationID.String()), + ) + + return err + } + + return nil +} + +// claimNextSigningGroup claims the next (organization, signatory) group whose +// signatures are due and returns the documents to list. claimStatusRaced means +// another worker claimed the candidate group first and the caller should retry. +func (h *documentNotificationHandler) claimNextSigningGroup( + ctx context.Context, + now time.Time, + debounceBefore time.Time, +) (documentNotificationTask, claimStatus, error) { + var ( + signatures coredata.DocumentVersionSignatures + claimed []gid.GID + ) + + if err := h.service.pg.WithTx( + ctx, + func(ctx context.Context, tx pg.Tx) error { + if err := signatures.LoadNextDueGroupForNotification(ctx, tx, now, debounceBefore, h.reminderInterval); err != nil { + return fmt.Errorf("cannot load next due signature group: %w", err) + } + + if len(signatures) == 0 { + return nil + } + + dueClaimed, err := signatures.ClaimForNotification(ctx, tx, now, debounceBefore, h.reminderInterval) + if err != nil { + return fmt.Errorf("cannot claim signatures for notification: %w", err) + } + + if len(dueClaimed) == 0 { + return nil + } + + bumped, err := signatures.BumpRemainingForNotification(ctx, tx, dueClaimed, now) + if err != nil { + return fmt.Errorf("cannot bump remaining signatures for notification: %w", err) + } + + claimed = append(dueClaimed, bumped...) + + return nil + }, + ); err != nil { + return documentNotificationTask{}, claimStatusNone, err + } + + if len(signatures) == 0 { + return documentNotificationTask{}, claimStatusNone, nil + } + + if len(claimed) == 0 { + return documentNotificationTask{}, claimStatusRaced, nil + } + + claimedSet := make(map[gid.GID]struct{}, len(claimed)) + for _, id := range claimed { + claimedSet[id] = struct{}{} + } + + versionIDs := make([]gid.GID, 0, len(signatures)) + for _, signature := range signatures { + if _, ok := claimedSet[signature.ID]; ok { + versionIDs = append(versionIDs, signature.DocumentVersionID) + } + } + + group := signatures[0] + + return documentNotificationTask{ + kind: notificationKindSigning, + organizationID: group.OrganizationID, + recipientID: group.SignedBy, + versionIDs: versionIDs, + }, claimStatusClaimed, nil +} + +// claimNextApprovalGroup claims the next (organization, approver) group whose +// decisions are due and resolves the documents to list via their quorums. +// claimStatusRaced means another worker won the candidate group first. +func (h *documentNotificationHandler) claimNextApprovalGroup( + ctx context.Context, + now time.Time, + debounceBefore time.Time, +) (documentNotificationTask, claimStatus, error) { + var ( + decisions coredata.DocumentVersionApprovalDecisions + claimed []gid.GID + versionIDs []gid.GID + organizationID gid.GID + recipientID gid.GID + ) + + if err := h.service.pg.WithTx( + ctx, + func(ctx context.Context, tx pg.Tx) error { + if err := decisions.LoadNextDueGroupForNotification(ctx, tx, now, debounceBefore, h.reminderInterval); err != nil { + return fmt.Errorf("cannot load next due approval group: %w", err) + } + + if len(decisions) == 0 { + return nil + } + + dueClaimed, err := decisions.ClaimForNotification(ctx, tx, now, debounceBefore, h.reminderInterval) + if err != nil { + return fmt.Errorf("cannot claim approval decisions for notification: %w", err) + } + + if len(dueClaimed) == 0 { + return nil + } + + bumped, err := decisions.BumpRemainingForNotification(ctx, tx, dueClaimed, now) + if err != nil { + return fmt.Errorf("cannot bump remaining approval decisions for notification: %w", err) + } + + claimed = append(dueClaimed, bumped...) + + claimedSet := make(map[gid.GID]struct{}, len(claimed)) + for _, id := range claimed { + claimedSet[id] = struct{}{} + } + + quorumIDs := make([]gid.GID, 0, len(claimed)) + + for _, decision := range decisions { + if _, ok := claimedSet[decision.ID]; ok { + quorumIDs = append(quorumIDs, decision.QuorumID) + } + } + + scope := coredata.NewScopeFromObjectID(decisions[0].OrganizationID) + + var quorums coredata.DocumentVersionApprovalQuorums + if err := quorums.LoadByIDs(ctx, tx, scope, quorumIDs); err != nil { + return fmt.Errorf("cannot load approval quorums: %w", err) + } + + for _, quorum := range quorums { + versionIDs = append(versionIDs, quorum.VersionID) + } + + organizationID = decisions[0].OrganizationID + recipientID = decisions[0].ApproverID + + return nil + }, + ); err != nil { + return documentNotificationTask{}, claimStatusNone, err + } + + if len(decisions) == 0 { + return documentNotificationTask{}, claimStatusNone, nil + } + + if len(claimed) == 0 { + return documentNotificationTask{}, claimStatusRaced, nil + } + + return documentNotificationTask{ + kind: notificationKindApproval, + organizationID: organizationID, + recipientID: recipientID, + versionIDs: versionIDs, + }, claimStatusClaimed, nil +} + +func (s *DocumentService) sendNotification( + ctx context.Context, + tx pg.Tx, + scope coredata.Scoper, + kind notificationKind, + organizationID gid.GID, + recipientID gid.GID, + versionIDs []gid.GID, +) error { + if len(versionIDs) == 0 { + return nil + } + + var versions coredata.DocumentVersions + if err := versions.LoadByIDs(ctx, tx, scope, versionIDs); err != nil { + return fmt.Errorf("cannot load document versions for notification: %w", err) + } + + if len(versions) == 0 { + return nil + } + + var profiles coredata.MembershipProfiles + if err := profiles.LoadByIDs(ctx, tx, scope, []gid.GID{recipientID}); err != nil { + return fmt.Errorf("cannot load notification recipient: %w", err) + } + + if len(profiles) == 0 { + return nil + } + + recipient := profiles[0] + + organization := &coredata.Organization{} + if err := organization.LoadByID(ctx, tx, scope, organizationID); err != nil { + return fmt.Errorf("cannot load notification organization: %w", err) + } + + token, err := s.buildInvitationToken(ctx, tx, recipient) + if err != nil { + return err + } + + documents := make([]emails.DocumentSummary, 0, len(versions)) + for _, version := range versions { + documentPath, err := documentDestinationPath(kind, organizationID, version.DocumentID) + if err != nil { + return fmt.Errorf("cannot build document destination path: %w", err) + } + + documentURL, err := s.recipientURL(documentPath, token) + if err != nil { + return fmt.Errorf("cannot build document notification URL: %w", err) + } + + documents = append(documents, emails.DocumentSummary{ + Title: version.Title, + Type: version.DocumentType.Label(), + URL: documentURL, + }) + } + + sort.Slice(documents, func(i, j int) bool { + return documents[i].Title < documents[j].Title + }) + + mainPath, err := mainDestinationPath(kind, organizationID) + if err != nil { + return fmt.Errorf("cannot build main destination path: %w", err) + } + + mainURL, err := s.recipientURL(mainPath, token) + if err != nil { + return fmt.Errorf("cannot build main notification URL: %w", err) + } + + email, err := s.renderNotificationEmail(ctx, kind, recipient, organization.Name, mainURL, documents) + if err != nil { + return fmt.Errorf("cannot render notification email: %w", err) + } + + if err := email.Insert(ctx, tx); err != nil { + return fmt.Errorf("cannot insert notification email: %w", err) + } + + return nil +} + +func (s *DocumentService) renderNotificationEmail( + ctx context.Context, + kind notificationKind, + recipient *coredata.MembershipProfile, + organizationName string, + mainURL string, + documents []emails.DocumentSummary, +) (*coredata.Email, error) { + emailPresenter := emails.NewPresenter(s.svc.baseURL, recipient.FullName) + + var ( + subject string + textBody string + htmlBody *string + err error + ) + + switch kind { + case notificationKindSigning: + subject, textBody, htmlBody, err = emailPresenter.RenderDocumentSigning(ctx, mainURL, organizationName, documents) + case notificationKindApproval: + subject, textBody, htmlBody, err = emailPresenter.RenderDocumentApproval(ctx, mainURL, organizationName, documents) + default: + return nil, fmt.Errorf("unknown notification kind %q", kind) + } + + if err != nil { + return nil, fmt.Errorf("cannot render notification email body: %w", err) + } + + return coredata.NewEmail( + recipient.FullName, + recipient.EmailAddress, + subject, + textBody, + htmlBody, + &coredata.EmailOptions{ + SenderName: new(organizationName), + }, + ), nil +} + +// buildInvitationToken returns an activation token for invited recipients that +// are not yet active and not managed by SCIM; others get an empty token and a +// direct link. +func (s *DocumentService) buildInvitationToken( + ctx context.Context, + tx pg.Tx, + recipient *coredata.MembershipProfile, +) (string, error) { + if recipient.State == coredata.ProfileStateActive || recipient.Source == coredata.ProfileSourceSCIM { + return "", nil + } + + now := time.Now() + invitation := &coredata.Invitation{ + ID: gid.New(recipient.OrganizationID.TenantID(), coredata.InvitationEntityType), + OrganizationID: recipient.OrganizationID, + UserID: recipient.ID, + Status: coredata.InvitationStatusPending, + ExpiresAt: now.Add(s.invitationTokenValidity), + CreatedAt: now, + } + + if err := invitation.Insert(ctx, tx, coredata.NewScopeFromObjectID(recipient.OrganizationID)); err != nil { + return "", fmt.Errorf("cannot insert invitation: %w", err) + } + + invitationToken, err := statelesstoken.NewToken( + s.tokenSecret, + iam.TokenTypeOrganizationInvitation, + s.invitationTokenValidity, + iam.InvitationTokenData{InvitationID: invitation.ID}, + ) + if err != nil { + return "", fmt.Errorf("cannot generate invitation token: %w", err) + } + + return invitationToken, nil +} + +// recipientURL builds the absolute link for destinationPath, routing through the +// account activation flow when token is set. +func (s *DocumentService) recipientURL(destinationPath string, token string) (string, error) { + target, err := baseurl.MustParse(s.svc.baseURL).AppendPath(destinationPath).String() + if err != nil { + return "", fmt.Errorf("cannot build destination URL: %w", err) + } + + if token == "" { + return target, nil + } + + activationURL, err := baseurl.MustParse(s.svc.baseURL). + AppendPath("/auth/activate-account"). + WithQuery("token", token). + WithQuery("continue", target). + String() + if err != nil { + return "", fmt.Errorf("cannot build activation URL: %w", err) + } + + return activationURL, nil +} + +func mainDestinationPath(kind notificationKind, organizationID gid.GID) (string, error) { + path, err := url.JoinPath("/organizations", organizationID.String(), "employee", notificationSection(kind)) + if err != nil { + return "", fmt.Errorf("cannot build notification path: %w", err) + } + + return path, nil +} + +func documentDestinationPath(kind notificationKind, organizationID gid.GID, documentID gid.GID) (string, error) { + path, err := url.JoinPath("/organizations", organizationID.String(), "employee", notificationSection(kind), documentID.String()) + if err != nil { + return "", fmt.Errorf("cannot build document notification path: %w", err) + } + + return path, nil +} + +func notificationSection(kind notificationKind) string { + if kind == notificationKindApproval { + return "approvals" + } + + return "signatures" +} diff --git a/pkg/probo/document_service.go b/pkg/probo/document_service.go index 38c9d6c8a..d86a02053 100644 --- a/pkg/probo/document_service.go +++ b/pkg/probo/document_service.go @@ -23,7 +23,6 @@ import ( "errors" "fmt" "io" - "net/url" "os" "regexp" "strings" @@ -35,19 +34,16 @@ import ( "go.gearno.de/kit/pg" "go.probo.inc/probo/packages/emails" "go.probo.inc/probo/pkg/agent" - "go.probo.inc/probo/pkg/baseurl" "go.probo.inc/probo/pkg/coredata" "go.probo.inc/probo/pkg/docgen" "go.probo.inc/probo/pkg/esign" "go.probo.inc/probo/pkg/gid" "go.probo.inc/probo/pkg/html2pdf" - "go.probo.inc/probo/pkg/iam" "go.probo.inc/probo/pkg/llm" "go.probo.inc/probo/pkg/mail" "go.probo.inc/probo/pkg/page" "go.probo.inc/probo/pkg/pdfutils" "go.probo.inc/probo/pkg/prosemirror" - "go.probo.inc/probo/pkg/statelesstoken" "go.probo.inc/probo/pkg/validator" ) @@ -756,102 +752,6 @@ func (s *DocumentService) Create( return document, documentVersion, nil } -func (s *DocumentService) SendSigningNotifications( - ctx context.Context, scope coredata.Scoper, - organizationID gid.GID, -) error { - now := time.Now() - - err := s.svc.pg.WithTx( - ctx, - func(ctx context.Context, tx pg.Tx) error { - var signatories coredata.MembershipProfiles - if err := signatories.LoadAwaitingSigning(ctx, tx, scope); err != nil { - return fmt.Errorf("cannot load signatories: %w", err) - } - - organization := &coredata.Organization{} - if err := organization.LoadByID(ctx, tx, scope, organizationID); err != nil { - return fmt.Errorf("cannot load organization: %w", err) - } - - for _, signatory := range signatories { - emailPresenter := emails.NewPresenter(s.svc.baseURL, signatory.FullName) - - var ( - employeeDocumentsURLPath = "/organizations/" + organizationID.String() + "/employee" - emailLinkURLPath = employeeDocumentsURLPath - query = make(url.Values) - ) - - if signatory.State != coredata.ProfileStateActive { - if signatory.Source != coredata.ProfileSourceSCIM { - invitation := &coredata.Invitation{ - ID: gid.New(organizationID.TenantID(), coredata.InvitationEntityType), - OrganizationID: organizationID, - UserID: signatory.ID, - Status: coredata.InvitationStatusPending, - ExpiresAt: now.Add(s.invitationTokenValidity), - CreatedAt: now, - } - if err := invitation.Insert(ctx, tx, coredata.NewScopeFromObjectID(organizationID)); err != nil { - return fmt.Errorf("cannot insert invitation: %w", err) - } - - invitationToken, err := statelesstoken.NewToken( - s.tokenSecret, - iam.TokenTypeOrganizationInvitation, - s.invitationTokenValidity, - iam.InvitationTokenData{InvitationID: invitation.ID}, - ) - if err != nil { - return fmt.Errorf("cannot generate invitation token: %w", err) - } - - emailLinkURLPath = "/auth/activate-account" - continueURL := baseurl.MustParse(s.svc.baseURL).AppendPath(employeeDocumentsURLPath).MustString() - - query.Add("token", invitationToken) - query.Add("continue", continueURL) - } - } - - subject, textBody, htmlBody, err := emailPresenter.RenderDocumentSigning( - ctx, - emailLinkURLPath, - query, - organization.Name, - ) - if err != nil { - return fmt.Errorf("cannot render signing request email: %w", err) - } - - email := coredata.NewEmail( - signatory.FullName, - signatory.EmailAddress, - subject, - textBody, - htmlBody, - &coredata.EmailOptions{ - SenderName: new(organization.Name), - }, - ) - - if err := email.Insert(ctx, tx); err != nil { - return fmt.Errorf("cannot insert email: %w", err) - } - } - - return nil - }, - ) - if err != nil { - return fmt.Errorf("cannot send signing notifications: %w", err) - } - - return nil -} - func (s *DocumentService) SignDocumentVersionByIdentity( ctx context.Context, scope coredata.Scoper, req SignDocumentVersionRequest, diff --git a/pkg/probod/aliases.go b/pkg/probod/aliases.go index ae28af29d..e1028ffbe 100644 --- a/pkg/probod/aliases.go +++ b/pkg/probod/aliases.go @@ -51,9 +51,11 @@ type ( SMTPConfig = probodconfig.SMTPConfig NotificationsConfig = probodconfig.NotificationsConfig WebhookConfig = probodconfig.WebhookConfig - OIDCProviderConfig = probodconfig.OIDCProviderConfig - PgConfig = probodconfig.PgConfig - SAMLConfig = probodconfig.SAMLConfig - SCIMBridgeConfig = probodconfig.SCIMBridgeConfig - SlackConfig = probodconfig.SlackConfig + + DocumentNotificationConfig = probodconfig.DocumentNotificationConfig + OIDCProviderConfig = probodconfig.OIDCProviderConfig + PgConfig = probodconfig.PgConfig + SAMLConfig = probodconfig.SAMLConfig + SCIMBridgeConfig = probodconfig.SCIMBridgeConfig + SlackConfig = probodconfig.SlackConfig ) diff --git a/pkg/probod/probod.go b/pkg/probod/probod.go index 879c0745b..36814f65b 100644 --- a/pkg/probod/probod.go +++ b/pkg/probod/probod.go @@ -152,6 +152,11 @@ func New() *Implm { SenderInterval: 5, CacheTTL: 86400, }, + Document: DocumentNotificationConfig{ + Interval: 300, // 5 minutes + DebounceDelay: 900, // 15 minutes + ReminderInterval: 86400, // 1 day base cadence (1x, 2x, 3x) + }, }, CustomDomains: CustomDomainsConfig{ RenewalInterval: 3600, @@ -734,6 +739,40 @@ func (impl *Implm) Run( }, ) + documentNotificationInterval := time.Duration(impl.cfg.Notifications.Document.Interval) * time.Second + if documentNotificationInterval <= 0 { + documentNotificationInterval = 5 * time.Minute + } + + documentNotificationDebounce := time.Duration(impl.cfg.Notifications.Document.DebounceDelay) * time.Second + if documentNotificationDebounce <= 0 { + documentNotificationDebounce = 15 * time.Minute + } + + documentNotificationReminder := time.Duration(impl.cfg.Notifications.Document.ReminderInterval) * time.Second + if documentNotificationReminder <= 0 { + documentNotificationReminder = 24 * time.Hour + } + + documentNotificationWorker := probo.NewDocumentNotificationWorker( + proboService, + l.Named("document-notification-worker"), + probo.DocumentNotificationWorkerConfig{ + DebounceDelay: documentNotificationDebounce, + ReminderInterval: documentNotificationReminder, + }, + worker.WithInterval(documentNotificationInterval), + ) + documentNotificationCtx, stopDocumentNotification := context.WithCancel(context.Background()) + + wg.Go( + func() { + if err := documentNotificationWorker.Run(documentNotificationCtx); err != nil { + cancel(fmt.Errorf("document notification worker crashed: %w", err)) + } + }, + ) + accessReviewWorkerCtx, stopAccessReviewWorker := context.WithCancel(context.Background()) wg.Go( @@ -959,6 +998,7 @@ func (impl *Implm) Run( stopVettingWorker() stopEvidenceDescriptionWorker() stopDocumentPDFWorker() + stopDocumentNotification() stopExportJobExporter() stopAccessReviewWorker() stopIAMService() diff --git a/pkg/probodconfig/notifications_config.go b/pkg/probodconfig/notifications_config.go index be7c89c4b..24675072d 100644 --- a/pkg/probodconfig/notifications_config.go +++ b/pkg/probodconfig/notifications_config.go @@ -15,12 +15,26 @@ package probodconfig type NotificationsConfig struct { - Mailer MailerConfig `json:"mailer"` - Slack SlackConfig `json:"slack"` - Webhook WebhookConfig `json:"webhook"` + Mailer MailerConfig `json:"mailer"` + Slack SlackConfig `json:"slack"` + Webhook WebhookConfig `json:"webhook"` + Document DocumentNotificationConfig `json:"document"` } type WebhookConfig struct { SenderInterval int `json:"sender-interval"` CacheTTL int `json:"cache-ttl"` } + +// DocumentNotificationConfig configures the debounced worker that batches +// signature and approval request notifications. All durations are in seconds. +type DocumentNotificationConfig struct { + // Interval is how often the worker scans for pending requests. + Interval int `json:"interval"` + // DebounceDelay is how long a request must have been pending before its + // first notification is sent. + DebounceDelay int `json:"debounce-delay"` + // ReminderInterval is the base reminder cadence. Reminders are sent at + // 1x, 2x and 3x this interval after the previous email, then stop. + ReminderInterval int `json:"reminder-interval"` +} diff --git a/pkg/server/api/console/v1/document_resolvers.go b/pkg/server/api/console/v1/document_resolvers.go index 95d743ea7..610fece43 100644 --- a/pkg/server/api/console/v1/document_resolvers.go +++ b/pkg/server/api/console/v1/document_resolvers.go @@ -1386,23 +1386,6 @@ func (r *mutationResolver) BulkRequestSignatures(ctx context.Context, input type }, nil } -// SendSigningNotifications is the resolver for the sendSigningNotifications field. -func (r *mutationResolver) SendSigningNotifications(ctx context.Context, input types.SendSigningNotificationsInput) (*types.SendSigningNotificationsPayload, error) { - scope, err := r.authorize(ctx, input.OrganizationID, probo.ActionDocumentSendSigningNotifications) - if err != nil { - return nil, err - } - - if err := r.probo.Documents.SendSigningNotifications(ctx, scope, input.OrganizationID); err != nil { - r.logger.ErrorCtx(ctx, "cannot send signing notifications", log.Error(err)) - return nil, gqlutils.Internal(ctx) - } - - return &types.SendSigningNotificationsPayload{ - Success: true, - }, nil -} - // CancelSignatureRequest is the resolver for the cancelSignatureRequest field. func (r *mutationResolver) CancelSignatureRequest(ctx context.Context, input types.CancelSignatureRequestInput) (*types.CancelSignatureRequestPayload, error) { scope, err := r.authorize(ctx, input.DocumentVersionSignatureID, probo.ActionDocumentVersionCancelSignature) diff --git a/pkg/server/api/console/v1/graphql/document.graphql b/pkg/server/api/console/v1/graphql/document.graphql index cd31bc7b8..988eacfdb 100644 --- a/pkg/server/api/console/v1/graphql/document.graphql +++ b/pkg/server/api/console/v1/graphql/document.graphql @@ -567,9 +567,6 @@ extend type Mutation { bulkRequestSignatures( input: BulkRequestSignaturesInput! ): BulkRequestSignaturesPayload! - sendSigningNotifications( - input: SendSigningNotificationsInput! - ): SendSigningNotificationsPayload! cancelSignatureRequest( input: CancelSignatureRequestInput! ): CancelSignatureRequestPayload! @@ -685,10 +682,6 @@ input BulkRequestSignaturesInput { signatoryIds: [ID!]! } -input SendSigningNotificationsInput { - organizationId: ID! -} - input CancelSignatureRequestInput { documentVersionSignatureId: ID! } @@ -782,10 +775,6 @@ type BulkRequestSignaturesPayload { documentVersionSignatureEdges: [DocumentVersionSignatureEdge!]! } -type SendSigningNotificationsPayload { - success: Boolean! -} - type CancelSignatureRequestPayload { deletedDocumentVersionSignatureId: ID! } diff --git a/pkg/server/api/mcp/v1/schema.resolvers.go b/pkg/server/api/mcp/v1/schema.resolvers.go index 0e54cbdc8..e6399d180 100644 --- a/pkg/server/api/mcp/v1/schema.resolvers.go +++ b/pkg/server/api/mcp/v1/schema.resolvers.go @@ -4125,24 +4125,6 @@ func (r *Resolver) VoidDocumentVersionApprovalTool(ctx context.Context, req *mcp }, nil } -func (r *Resolver) SendSigningNotificationsTool(ctx context.Context, req *mcp.CallToolRequest, input *types.SendSigningNotificationsInput) (*mcp.CallToolResult, types.SendSigningNotificationsOutput, error) { - scope, err := r.Authorize(ctx, input.OrganizationID, probo.ActionDocumentSendSigningNotifications) - if err != nil { - return nil, types.SendSigningNotificationsOutput{}, err - } - - svc := r.proboSvc - - err = svc.Documents.SendSigningNotifications(ctx, scope, input.OrganizationID) - if err != nil { - panic(fmt.Errorf("cannot send signing notifications: %w", err)) - } - - return nil, types.SendSigningNotificationsOutput{ - Success: true, - }, nil -} - func (r *Resolver) DeleteDocumentDraftTool(ctx context.Context, req *mcp.CallToolRequest, input *types.DeleteDocumentDraftInput) (*mcp.CallToolResult, types.DeleteDocumentDraftOutput, error) { scope, err := r.Authorize(ctx, input.ID, probo.ActionDocumentDeleteDraft) if err != nil { diff --git a/pkg/server/api/mcp/v1/specification.yaml b/pkg/server/api/mcp/v1/specification.yaml index 6286e8200..a925e9629 100644 --- a/pkg/server/api/mcp/v1/specification.yaml +++ b/pkg/server/api/mcp/v1/specification.yaml @@ -6570,24 +6570,6 @@ components: approval_decision: $ref: "#/components/schemas/DocumentVersionApprovalDecision" - SendSigningNotificationsInput: - type: object - required: - - organization_id - properties: - organization_id: - $ref: "#/components/schemas/GID" - description: Organization ID - - SendSigningNotificationsOutput: - type: object - required: - - success - properties: - success: - type: boolean - description: Whether the notifications were sent successfully - WebhookEventType: type: string enum: @@ -13109,14 +13091,6 @@ tools: $ref: "#/components/schemas/VoidDocumentVersionApprovalInput" outputSchema: $ref: "#/components/schemas/VoidDocumentVersionApprovalOutput" - - name: sendSigningNotifications - description: Send signing notifications to all signatories with pending signature requests in the organization - hints: - readonly: false - inputSchema: - $ref: "#/components/schemas/SendSigningNotificationsInput" - outputSchema: - $ref: "#/components/schemas/SendSigningNotificationsOutput" - name: listStatementsOfApplicability description: List all statements of applicability for the organization hints: