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: