rohit/probo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add processing activity, DPIA and TIA publish to document system

Browse Source 
Replace the old PDF/snapshot-based exports for processing activities,
Data Protection Impact Assessments and Transfer Impact Assessments with
the publish document system. Includes GraphQL mutations, MCP tools, CLI
commands, n8n operations, frontend publish dialogs, e2e tests, and
prosemirror register templates that mirror the previous PDF layouts.

Each register lives as a generated DocumentTypeRegister document on the
organization, reused across publishes (the major version bumps on every
republish). Approvers can be passed in to create a draft pending
approval; otherwise the version is published immediately. The frontend
ProcessingActivities page exposes a Publish dropdown per register and a
Document link button per active tab, pre-fills the previous default
approvers, and navigates to the published document on success.

Remove snapshot mode entirely from these three entities: drop snapshotId
and sourceId from GraphQL schemas, types, filters, resolvers, MCP spec,
frontend routes and pages; remove SnapshotsTypeProcessingActivities from
the snapshot registry and delete the ProcessingActivities.Snapshot,
ProcessingActivitySnapshotter interface and *.InsertProcessingActivitySnapshots
methods. The snapshot_id columns remain in the database but are now
filtered out with snapshot_id IS NULL.

Add Get/Upsert/Clear GeneratedDocumentID methods on each entity type
(ProcessingActivity, DataProtectionImpactAssessment,
TransferImpactAssessment) backed by new columns in the generated_documents
table, matching the Finding/Obligation pattern.

Signed-off-by: Sacha Al Himdani <sacha@getprobo.com>
This commit is contained in:
Sacha Al Himdani
2026-04-28 16:07:18 +02:00
parent be294599ec
commit ba8bce2ad3
60 changed files with 5461 additions and 3601 deletions
Download Patch File Download Diff File Expand all files Collapse all files

183
pkg/coredata/data_protection_impact_assessment.go
View File

@@ -28,6 +28,113 @@ import (
"go.probo.inc/probo/pkg/page"
)
func (d DataProtectionImpactAssessment) GetGeneratedDocumentID(
ctx context.Context,
conn pg.Querier,
organizationID gid.GID,
) (*gid.GID, error) {
var documentID *gid.GID
err := conn.QueryRow(
ctx,
`
SELECT
data_protection_impact_assessments_document_id
FROM
generated_documents
WHERE
organization_id = @organization_id
`,
pgx.NamedArgs{"organization_id": organizationID},
).Scan(&documentID)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("cannot get DPIA list document ID: %w", err)
}
return documentID, nil
}
func (d DataProtectionImpactAssessment) UpsertGeneratedDocumentID(
ctx context.Context,
conn pg.Tx,
organizationID gid.GID,
tenantID gid.TenantID,
documentID gid.GID,
) error {
now := time.Now()
_, err := conn.Exec(
ctx,
`
INSERT INTO generated_documents (
organization_id,
tenant_id,
data_protection_impact_assessments_document_id,
created_at,
updated_at
) VALUES (
@organization_id,
@tenant_id,
@data_protection_impact_assessments_document_id,
@created_at,
@updated_at
)
ON CONFLICT (organization_id) DO UPDATE
SET
data_protection_impact_assessments_document_id = @data_protection_impact_assessments_document_id,
updated_at = @updated_at
`,
pgx.NamedArgs{
"organization_id": organizationID,
"tenant_id": tenantID,
"data_protection_impact_assessments_document_id": documentID,
"created_at": now,
"updated_at": now,
},
)
if err != nil {
return fmt.Errorf("cannot upsert DPIA list document ID: %w", err)
}
return nil
}
func (d DataProtectionImpactAssessment) ClearGeneratedDocumentID(
ctx context.Context,
conn pg.Tx,
documentIDs []gid.GID,
) error {
ids := make([]string, len(documentIDs))
for i, id := range documentIDs {
ids[i] = id.String()
}
_, err := conn.Exec(
ctx,
`
UPDATE
generated_documents
SET
data_protection_impact_assessments_document_id = NULL,
updated_at = @now
WHERE
data_protection_impact_assessments_document_id = ANY(@ids)
`,
pgx.NamedArgs{
"ids": ids,
"now": time.Now(),
},
)
if err != nil {
return fmt.Errorf("cannot clear DPIA list document references: %w", err)
}
return nil
}
type (
DataProtectionImpactAssessment struct {
ID gid.GID `db:"id"`
@@ -76,7 +183,6 @@ func (dpias *DataProtectionImpactAssessments) CountByOrganizationID(
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
filter *DataProtectionImpactAssessmentFilter,
) (int, error) {
q := `
SELECT
@@ -86,14 +192,13 @@ FROM
WHERE
%s
AND organization_id = @organization_id
AND %s
AND snapshot_id IS NULL
`
q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment())
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{"organization_id": organizationID}
maps.Copy(args, scope.SQLArguments())
maps.Copy(args, filter.SQLArguments())
row := conn.QueryRow(ctx, q, args)
@@ -112,7 +217,6 @@ func (dpias *DataProtectionImpactAssessments) LoadByOrganizationID(
scope Scoper,
organizationID gid.GID,
cursor *page.Cursor[DataProtectionImpactAssessmentOrderField],
filter *DataProtectionImpactAssessmentFilter,
) error {
q := `
SELECT
@@ -133,15 +237,14 @@ FROM
WHERE
%s
AND organization_id = @organization_id
AND %s
AND snapshot_id IS NULL
AND %s
`
q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment(), cursor.SQLFragment())
q = fmt.Sprintf(q, scope.SQLFragment(), cursor.SQLFragment())
args := pgx.StrictNamedArgs{"organization_id": organizationID}
maps.Copy(args, scope.SQLArguments())
maps.Copy(args, filter.SQLArguments())
maps.Copy(args, cursor.SQLArguments())
rows, err := conn.Query(ctx, q, args)
@@ -164,7 +267,6 @@ func (dpias *DataProtectionImpactAssessments) LoadAllByOrganizationID(
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
filter *DataProtectionImpactAssessmentFilter,
) error {
q := `
SELECT
@@ -185,14 +287,13 @@ FROM
WHERE
%s
AND organization_id = @organization_id
AND %s
AND snapshot_id IS NULL
`
q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment())
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{"organization_id": organizationID}
maps.Copy(args, scope.SQLArguments())
maps.Copy(args, filter.SQLArguments())
rows, err := conn.Query(ctx, q, args)
if err != nil {
@@ -438,61 +539,3 @@ WHERE
return nil
}
func (dpias DataProtectionImpactAssessments) InsertProcessingActivitySnapshots(
ctx context.Context,
conn pg.Tx,
scope Scoper,
organizationID gid.GID,
snapshotID gid.GID,
) error {
query := `
INSERT INTO processing_activity_data_protection_impact_assessments (
id,
tenant_id,
snapshot_id,
source_id,
organization_id,
processing_activity_id,
description,
necessity_and_proportionality,
potential_risk,
mitigations,
residual_risk,
created_at,
updated_at
)
SELECT
generate_gid(decode_base64_unpadded(@tenant_id), @dpia_entity_type),
@tenant_id,
@snapshot_id,
dpia.id,
dpia.organization_id,
pa_snapshot.id,
dpia.description,
dpia.necessity_and_proportionality,
dpia.potential_risk,
dpia.mitigations,
dpia.residual_risk,
dpia.created_at,
dpia.updated_at
FROM processing_activity_data_protection_impact_assessments dpia
INNER JOIN processing_activities pa_source ON dpia.processing_activity_id = pa_source.id AND pa_source.snapshot_id IS NULL
INNER JOIN processing_activities pa_snapshot ON pa_source.id = pa_snapshot.source_id AND pa_snapshot.snapshot_id = @snapshot_id
WHERE dpia.tenant_id = @tenant_id AND dpia.organization_id = @organization_id AND dpia.snapshot_id IS NULL
`
args := pgx.StrictNamedArgs{
"tenant_id": scope.GetTenantID(),
"snapshot_id": snapshotID,
"organization_id": organizationID,
"dpia_entity_type": DataProtectionImpactAssessmentEntityType,
}
_, err := conn.Exec(ctx, query, args)
if err != nil {
return fmt.Errorf("cannot insert data protection impact assessment snapshots: %w", err)
}
return nil
}

61
pkg/coredata/data_protection_impact_assessment_filter.go
View File

@@ -1,61 +0,0 @@
// Copyright (c) 2025-2026 Probo Inc <hello@getprobo.com>.
//
// 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 coredata
import (
"github.com/jackc/pgx/v5"
"go.probo.inc/probo/pkg/gid"
)
type (
DataProtectionImpactAssessmentFilter struct {
snapshotID **gid.GID
}
)
func NewDataProtectionImpactAssessmentFilter(snapshotID **gid.GID) *DataProtectionImpactAssessmentFilter {
return &DataProtectionImpactAssessmentFilter{
snapshotID: snapshotID,
}
}
func (f *DataProtectionImpactAssessmentFilter) SQLArguments() pgx.NamedArgs {
args := pgx.NamedArgs{}
if f.snapshotID != nil && *f.snapshotID != nil {
args["filter_snapshot_id"] = **f.snapshotID
}
return args
}
func (f *DataProtectionImpactAssessmentFilter) SQLFragment() string {
if f.snapshotID == nil {
return "TRUE"
}
if *f.snapshotID == nil {
return "snapshot_id IS NULL"
} else {
return "snapshot_id = @filter_snapshot_id"
}
}
func (f *DataProtectionImpactAssessmentFilter) SnapshotID() *gid.GID {
if f.snapshotID == nil || *f.snapshotID == nil {
return nil
}
return *f.snapshotID
}

18
pkg/coredata/migrations/20260428T115900Z.sql Normal file
View File

@@ -0,0 +1,18 @@
-- Copyright (c) 2026 Probo Inc <hello@getprobo.com>.
--
-- 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 generated_documents
ADD COLUMN processing_activities_document_id TEXT REFERENCES documents(id) ON DELETE SET NULL,
ADD COLUMN data_protection_impact_assessments_document_id TEXT REFERENCES documents(id) ON DELETE SET NULL,
ADD COLUMN transfer_impact_assessments_document_id TEXT REFERENCES documents(id) ON DELETE SET NULL;

292
pkg/coredata/processing_activities.go
View File

@@ -27,6 +27,113 @@ import (
"go.probo.inc/probo/pkg/page"
)
func (p ProcessingActivity) GetGeneratedDocumentID(
ctx context.Context,
conn pg.Querier,
organizationID gid.GID,
) (*gid.GID, error) {
var documentID *gid.GID
err := conn.QueryRow(
ctx,
`
SELECT
processing_activities_document_id
FROM
generated_documents
WHERE
organization_id = @organization_id
`,
pgx.NamedArgs{"organization_id": organizationID},
).Scan(&documentID)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("cannot get processing activity list document ID: %w", err)
}
return documentID, nil
}
func (p ProcessingActivity) UpsertGeneratedDocumentID(
ctx context.Context,
conn pg.Tx,
organizationID gid.GID,
tenantID gid.TenantID,
documentID gid.GID,
) error {
now := time.Now()
_, err := conn.Exec(
ctx,
`
INSERT INTO generated_documents (
organization_id,
tenant_id,
processing_activities_document_id,
created_at,
updated_at
) VALUES (
@organization_id,
@tenant_id,
@processing_activities_document_id,
@created_at,
@updated_at
)
ON CONFLICT (organization_id) DO UPDATE
SET
processing_activities_document_id = @processing_activities_document_id,
updated_at = @updated_at
`,
pgx.NamedArgs{
"organization_id": organizationID,
"tenant_id": tenantID,
"processing_activities_document_id": documentID,
"created_at": now,
"updated_at": now,
},
)
if err != nil {
return fmt.Errorf("cannot upsert processing activity list document ID: %w", err)
}
return nil
}
func (p ProcessingActivity) ClearGeneratedDocumentID(
ctx context.Context,
conn pg.Tx,
documentIDs []gid.GID,
) error {
ids := make([]string, len(documentIDs))
for i, id := range documentIDs {
ids[i] = id.String()
}
_, err := conn.Exec(
ctx,
`
UPDATE
generated_documents
SET
processing_activities_document_id = NULL,
updated_at = @now
WHERE
processing_activities_document_id = ANY(@ids)
`,
pgx.NamedArgs{
"ids": ids,
"now": time.Now(),
},
)
if err != nil {
return fmt.Errorf("cannot clear processing activity list document references: %w", err)
}
return nil
}
type (
ProcessingActivity struct {
ID gid.GID `db:"id"`
@@ -150,7 +257,6 @@ func (p *ProcessingActivities) CountByOrganizationID(
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
filter *ProcessingActivityFilter,
) (int, error) {
q := `
SELECT
@@ -160,14 +266,13 @@ FROM
WHERE
%s
AND organization_id = @organization_id
AND %s
AND snapshot_id IS NULL
`
q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment())
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{"organization_id": organizationID}
maps.Copy(args, scope.SQLArguments())
maps.Copy(args, filter.SQLArguments())
row := conn.QueryRow(ctx, q, args)
@@ -180,13 +285,82 @@ WHERE
return count, nil
}
func (p *ProcessingActivities) LoadByIDs(
ctx context.Context,
conn pg.Querier,
scope Scoper,
processingActivityIDs []gid.GID,
) error {
if len(processingActivityIDs) == 0 {
*p = ProcessingActivities{}
return nil
}
q := `
SELECT
id,
snapshot_id,
source_id,
organization_id,
name,
purpose,
data_subject_category,
personal_data_category,
special_or_criminal_data,
consent_evidence_link,
lawful_basis,
recipients,
location,
international_transfers,
transfer_safeguards,
retention_period,
security_measures,
data_protection_impact_assessment_needed,
transfer_impact_assessment_needed,
last_review_date,
next_review_date,
role,
dpo_profile_id,
created_at,
updated_at
FROM
processing_activities
WHERE
%s
AND id = ANY(@ids)
`
q = fmt.Sprintf(q, scope.SQLFragment())
ids := make([]string, len(processingActivityIDs))
for i, id := range processingActivityIDs {
ids[i] = id.String()
}
args := pgx.StrictNamedArgs{"ids": ids}
maps.Copy(args, scope.SQLArguments())
rows, err := conn.Query(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot query processing activities: %w", err)
}
processingActivities, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[ProcessingActivity])
if err != nil {
return fmt.Errorf("cannot collect processing activities: %w", err)
}
*p = processingActivities
return nil
}
func (p *ProcessingActivities) LoadByOrganizationID(
ctx context.Context,
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
cursor *page.Cursor[ProcessingActivityOrderField],
filter *ProcessingActivityFilter,
) error {
q := `
SELECT
@@ -220,15 +394,14 @@ FROM
WHERE
%s
AND organization_id = @organization_id
AND %s
AND snapshot_id IS NULL
AND %s
`
q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment(), cursor.SQLFragment())
q = fmt.Sprintf(q, scope.SQLFragment(), cursor.SQLFragment())
args := pgx.StrictNamedArgs{"organization_id": organizationID}
maps.Copy(args, scope.SQLArguments())
maps.Copy(args, filter.SQLArguments())
maps.Copy(args, cursor.SQLArguments())
rows, err := conn.Query(ctx, q, args)
@@ -251,7 +424,6 @@ func (p *ProcessingActivities) LoadAllByOrganizationID(
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
filter *ProcessingActivityFilter,
) error {
q := `
SELECT
@@ -285,15 +457,14 @@ FROM
WHERE
%s
AND organization_id = @organization_id
AND %s
AND snapshot_id IS NULL
ORDER BY created_at DESC
`
q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment())
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{"organization_id": organizationID}
maps.Copy(args, scope.SQLArguments())
maps.Copy(args, filter.SQLArguments())
rows, err := conn.Query(ctx, q, args)
if err != nil {
@@ -504,100 +675,3 @@ WHERE
return nil
}
func (pas ProcessingActivities) Snapshot(ctx context.Context, conn pg.Tx, scope Scoper, organizationID, snapshotID gid.GID) error {
snapshotters := []ProcessingActivitySnapshotter{ProcessingActivities{}, Vendors{}, ProcessingActivityVendors{}, DataProtectionImpactAssessments{}, TransferImpactAssessments{}}
for _, snapshotter := range snapshotters {
if err := snapshotter.InsertProcessingActivitySnapshots(ctx, conn, scope, organizationID, snapshotID); err != nil {
return fmt.Errorf("cannot create processing activity snapshots: (%T) %w", snapshotter, err)
}
}
return nil
}
func (pas ProcessingActivities) InsertProcessingActivitySnapshots(
ctx context.Context,
conn pg.Tx,
scope Scoper,
organizationID gid.GID,
snapshotID gid.GID,
) error {
query := `
INSERT INTO processing_activities (
id,
tenant_id,
snapshot_id,
source_id,
organization_id,
name,
purpose,
data_subject_category,
personal_data_category,
special_or_criminal_data,
consent_evidence_link,
lawful_basis,
recipients,
location,
international_transfers,
transfer_safeguards,
retention_period,
security_measures,
data_protection_impact_assessment_needed,
transfer_impact_assessment_needed,
last_review_date,
next_review_date,
role,
dpo_profile_id,
created_at,
updated_at
)
SELECT
generate_gid(decode_base64_unpadded(@tenant_id), @processing_activity_entity_type),
@tenant_id,
@snapshot_id,
par.id,
par.organization_id,
par.name,
par.purpose,
par.data_subject_category,
par.personal_data_category,
par.special_or_criminal_data,
par.consent_evidence_link,
par.lawful_basis,
par.recipients,
par.location,
par.international_transfers,
par.transfer_safeguards,
par.retention_period,
par.security_measures,
par.data_protection_impact_assessment_needed,
par.transfer_impact_assessment_needed,
par.last_review_date,
par.next_review_date,
par.role,
par.dpo_profile_id,
par.created_at,
par.updated_at
FROM processing_activities par
WHERE %s AND par.organization_id = @organization_id AND par.snapshot_id IS NULL
`
query = fmt.Sprintf(query, scope.SQLFragment())
args := pgx.StrictNamedArgs{
"tenant_id": scope.GetTenantID(),
"snapshot_id": snapshotID,
"organization_id": organizationID,
"processing_activity_entity_type": ProcessingActivityEntityType,
}
maps.Copy(args, scope.SQLArguments())
_, err := conn.Exec(ctx, query, args)
if err != nil {
return fmt.Errorf("cannot insert processing activity snapshots: %w", err)
}
return nil
}

61
pkg/coredata/processing_activity_filter.go
View File

@@ -1,61 +0,0 @@
// Copyright (c) 2025-2026 Probo Inc <hello@getprobo.com>.
//
// 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 coredata
import (
"github.com/jackc/pgx/v5"
"go.probo.inc/probo/pkg/gid"
)
type (
ProcessingActivityFilter struct {
snapshotID **gid.GID
}
)
func NewProcessingActivityFilter(snapshotID **gid.GID) *ProcessingActivityFilter {
return &ProcessingActivityFilter{
snapshotID: snapshotID,
}
}
func (f *ProcessingActivityFilter) SQLArguments() pgx.NamedArgs {
args := pgx.NamedArgs{}
if f.snapshotID != nil && *f.snapshotID != nil {
args["filter_snapshot_id"] = **f.snapshotID
}
return args
}
func (f *ProcessingActivityFilter) SQLFragment() string {
if f.snapshotID == nil {
return "TRUE"
}
if *f.snapshotID == nil {
return "snapshot_id IS NULL"
} else {
return "snapshot_id = @filter_snapshot_id"
}
}
func (f *ProcessingActivityFilter) SnapshotID() *gid.GID {
if f.snapshotID == nil || *f.snapshotID == nil {
return nil
}
return *f.snapshotID
}

63
pkg/coredata/processing_activity_vendor.go
View File

@@ -17,7 +17,6 @@ package coredata
import (
"context"
"fmt"
"maps"
"time"
"github.com/jackc/pgx/v5"
@@ -35,10 +34,6 @@ type (
}
ProcessingActivityVendors []*ProcessingActivityVendor
ProcessingActivitySnapshotter interface {
InsertProcessingActivitySnapshots(ctx context.Context, conn pg.Tx, scope Scoper, organizationID, snapshotID gid.GID) error
}
)
func (pav ProcessingActivityVendors) Merge(
@@ -124,61 +119,3 @@ FROM vendor_ids
return nil
}
func (pav ProcessingActivityVendors) InsertProcessingActivitySnapshots(
ctx context.Context,
conn pg.Tx,
scope Scoper,
organizationID gid.GID,
snapshotID gid.GID,
) error {
query := `
WITH
source_processing_activities AS (
SELECT id
FROM processing_activities
WHERE organization_id = @organization_id AND snapshot_id IS NULL
),
snapshot_processing_activities AS (
SELECT id, source_id
FROM processing_activities
WHERE organization_id = @organization_id AND snapshot_id = @snapshot_id
),
snapshot_vendors AS (
SELECT id, source_id
FROM vendors
WHERE organization_id = @organization_id AND snapshot_id = @snapshot_id
),
source_processing_activity_vendors AS (
SELECT processing_activity_id, vendor_id, snapshot_id, created_at
FROM processing_activity_vendors
WHERE %s AND processing_activity_id = ANY(SELECT id FROM source_processing_activities) AND snapshot_id IS NULL
)
INSERT INTO processing_activity_vendors (tenant_id, processing_activity_id, vendor_id, organization_id, snapshot_id, created_at)
SELECT
@tenant_id,
spa.id,
sv.id,
@organization_id,
@snapshot_id,
pav.created_at
FROM source_processing_activity_vendors pav
JOIN snapshot_processing_activities spa ON spa.source_id = pav.processing_activity_id
JOIN snapshot_vendors sv ON sv.source_id = pav.vendor_id
`
query = fmt.Sprintf(query, scope.SQLFragment())
args := pgx.StrictNamedArgs{
"snapshot_id": snapshotID,
"organization_id": organizationID,
}
maps.Copy(args, scope.SQLArguments())
_, err := conn.Exec(ctx, query, args)
if err != nil {
return fmt.Errorf("cannot insert processing activity vendor snapshots: %w", err)
}
return nil
}

1
pkg/coredata/snapshots_type.go
View File

@@ -38,7 +38,6 @@ func SnapshotsTypes() []SnapshotsType {
return []SnapshotsType{
SnapshotsTypeRisks,
SnapshotsTypeVendors,
SnapshotsTypeProcessingActivities,
}
}

2
pkg/coredata/snapshottable.go
View File

@@ -30,8 +30,6 @@ func GetSnapshottable(snapshotType SnapshotsType) (Snapshottable, error) {
switch snapshotType {
case SnapshotsTypeRisks:
return Risks{}, nil
case SnapshotsTypeProcessingActivities:
return ProcessingActivities{}, nil
case SnapshotsTypeVendors:
return Vendors{}, nil
default:

183
pkg/coredata/transfer_impact_assessment.go
View File

@@ -28,6 +28,113 @@ import (
"go.probo.inc/probo/pkg/page"
)
func (t TransferImpactAssessment) GetGeneratedDocumentID(
ctx context.Context,
conn pg.Querier,
organizationID gid.GID,
) (*gid.GID, error) {
var documentID *gid.GID
err := conn.QueryRow(
ctx,
`
SELECT
transfer_impact_assessments_document_id
FROM
generated_documents
WHERE
organization_id = @organization_id
`,
pgx.NamedArgs{"organization_id": organizationID},
).Scan(&documentID)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("cannot get TIA list document ID: %w", err)
}
return documentID, nil
}
func (t TransferImpactAssessment) UpsertGeneratedDocumentID(
ctx context.Context,
conn pg.Tx,
organizationID gid.GID,
tenantID gid.TenantID,
documentID gid.GID,
) error {
now := time.Now()
_, err := conn.Exec(
ctx,
`
INSERT INTO generated_documents (
organization_id,
tenant_id,
transfer_impact_assessments_document_id,
created_at,
updated_at
) VALUES (
@organization_id,
@tenant_id,
@transfer_impact_assessments_document_id,
@created_at,
@updated_at
)
ON CONFLICT (organization_id) DO UPDATE
SET
transfer_impact_assessments_document_id = @transfer_impact_assessments_document_id,
updated_at = @updated_at
`,
pgx.NamedArgs{
"organization_id": organizationID,
"tenant_id": tenantID,
"transfer_impact_assessments_document_id": documentID,
"created_at": now,
"updated_at": now,
},
)
if err != nil {
return fmt.Errorf("cannot upsert TIA list document ID: %w", err)
}
return nil
}
func (t TransferImpactAssessment) ClearGeneratedDocumentID(
ctx context.Context,
conn pg.Tx,
documentIDs []gid.GID,
) error {
ids := make([]string, len(documentIDs))
for i, id := range documentIDs {
ids[i] = id.String()
}
_, err := conn.Exec(
ctx,
`
UPDATE
generated_documents
SET
transfer_impact_assessments_document_id = NULL,
updated_at = @now
WHERE
transfer_impact_assessments_document_id = ANY(@ids)
`,
pgx.NamedArgs{
"ids": ids,
"now": time.Now(),
},
)
if err != nil {
return fmt.Errorf("cannot clear TIA list document references: %w", err)
}
return nil
}
type (
TransferImpactAssessment struct {
ID gid.GID `db:"id"`
@@ -76,7 +183,6 @@ func (tias *TransferImpactAssessments) CountByOrganizationID(
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
filter *TransferImpactAssessmentFilter,
) (int, error) {
q := `
SELECT
@@ -86,14 +192,13 @@ FROM
WHERE
%s
AND organization_id = @organization_id
AND %s
AND snapshot_id IS NULL
`
q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment())
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{"organization_id": organizationID}
maps.Copy(args, scope.SQLArguments())
maps.Copy(args, filter.SQLArguments())
row := conn.QueryRow(ctx, q, args)
@@ -112,7 +217,6 @@ func (tias *TransferImpactAssessments) LoadByOrganizationID(
scope Scoper,
organizationID gid.GID,
cursor *page.Cursor[TransferImpactAssessmentOrderField],
filter *TransferImpactAssessmentFilter,
) error {
q := `
SELECT
@@ -133,15 +237,14 @@ FROM
WHERE
%s
AND organization_id = @organization_id
AND %s
AND snapshot_id IS NULL
AND %s
`
q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment(), cursor.SQLFragment())
q = fmt.Sprintf(q, scope.SQLFragment(), cursor.SQLFragment())
args := pgx.StrictNamedArgs{"organization_id": organizationID}
maps.Copy(args, scope.SQLArguments())
maps.Copy(args, filter.SQLArguments())
maps.Copy(args, cursor.SQLArguments())
rows, err := conn.Query(ctx, q, args)
@@ -164,7 +267,6 @@ func (tias *TransferImpactAssessments) LoadAllByOrganizationID(
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
filter *TransferImpactAssessmentFilter,
) error {
q := `
SELECT
@@ -185,14 +287,13 @@ FROM
WHERE
%s
AND organization_id = @organization_id
AND %s
AND snapshot_id IS NULL
`
q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment())
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{"organization_id": organizationID}
maps.Copy(args, scope.SQLArguments())
maps.Copy(args, filter.SQLArguments())
rows, err := conn.Query(ctx, q, args)
if err != nil {
@@ -436,61 +537,3 @@ WHERE
return nil
}
func (tias TransferImpactAssessments) InsertProcessingActivitySnapshots(
ctx context.Context,
conn pg.Tx,
scope Scoper,
organizationID gid.GID,
snapshotID gid.GID,
) error {
query := `
INSERT INTO processing_activity_transfer_impact_assessments (
id,
tenant_id,
snapshot_id,
source_id,
organization_id,
processing_activity_id,
data_subjects,
legal_mechanism,
transfer,
local_law_risk,
supplementary_measures,
created_at,
updated_at
)
SELECT
generate_gid(decode_base64_unpadded(@tenant_id), @tia_entity_type),
@tenant_id,
@snapshot_id,
tia.id,
tia.organization_id,
pa_snapshot.id,
tia.data_subjects,
tia.legal_mechanism,
tia.transfer,
tia.local_law_risk,
tia.supplementary_measures,
tia.created_at,
tia.updated_at
FROM processing_activity_transfer_impact_assessments tia
INNER JOIN processing_activities pa_source ON tia.processing_activity_id = pa_source.id AND pa_source.snapshot_id IS NULL
INNER JOIN processing_activities pa_snapshot ON pa_source.id = pa_snapshot.source_id AND pa_snapshot.snapshot_id = @snapshot_id
WHERE tia.tenant_id = @tenant_id AND tia.organization_id = @organization_id AND tia.snapshot_id IS NULL
`
args := pgx.StrictNamedArgs{
"tenant_id": scope.GetTenantID(),
"snapshot_id": snapshotID,
"organization_id": organizationID,
"tia_entity_type": TransferImpactAssessmentEntityType,
}
_, err := conn.Exec(ctx, query, args)
if err != nil {
return fmt.Errorf("cannot insert transfer impact assessment snapshots: %w", err)
}
return nil
}

61
pkg/coredata/transfer_impact_assessment_filter.go
View File

@@ -1,61 +0,0 @@
// Copyright (c) 2025-2026 Probo Inc <hello@getprobo.com>.
//
// 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 coredata
import (
"github.com/jackc/pgx/v5"
"go.probo.inc/probo/pkg/gid"
)
type (
TransferImpactAssessmentFilter struct {
snapshotID **gid.GID
}
)
func NewTransferImpactAssessmentFilter(snapshotID **gid.GID) *TransferImpactAssessmentFilter {
return &TransferImpactAssessmentFilter{
snapshotID: snapshotID,
}
}
func (f *TransferImpactAssessmentFilter) SQLArguments() pgx.NamedArgs {
args := pgx.NamedArgs{}
if f.snapshotID != nil && *f.snapshotID != nil {
args["filter_snapshot_id"] = **f.snapshotID
}
return args
}
func (f *TransferImpactAssessmentFilter) SQLFragment() string {
if f.snapshotID == nil {
return "TRUE"
}
if *f.snapshotID == nil {
return "snapshot_id IS NULL"
} else {
return "snapshot_id = @filter_snapshot_id"
}
}
func (f *TransferImpactAssessmentFilter) SnapshotID() *gid.GID {
if f.snapshotID == nil || *f.snapshotID == nil {
return nil
}
return *f.snapshotID
}

107
pkg/coredata/vendor.go
View File

@@ -1014,7 +1014,6 @@ func (v *Vendors) LoadAllByProcessingActivities(
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
filter *ProcessingActivityFilter,
) (map[gid.GID][]string, error) {
q := `
WITH filtered_processing_activities AS (
@@ -1025,7 +1024,7 @@ WITH filtered_processing_activities AS (
WHERE
pa.tenant_id = @tenant_id
AND pa.organization_id = @organization_id
AND %s
AND pa.snapshot_id IS NULL
),
filtered_vendors AS (
SELECT
@@ -1050,13 +1049,11 @@ WHERE
ORDER BY
pav.processing_activity_id, fv.name
`
q = fmt.Sprintf(q, filter.SQLFragment())
args := pgx.StrictNamedArgs{
"organization_id": organizationID,
}
maps.Copy(args, scope.SQLArguments())
maps.Copy(args, filter.SQLArguments())
rows, err := conn.Query(ctx, q, args)
if err != nil {
@@ -1173,108 +1170,6 @@ ORDER BY name ASC
return nil
}
func (vs Vendors) InsertProcessingActivitySnapshots(
ctx context.Context,
conn pg.Tx,
scope Scoper,
organizationID gid.GID,
snapshotID gid.GID,
) error {
query := `
WITH
source_processing_activities AS (
SELECT id
FROM processing_activities
WHERE organization_id = @organization_id AND snapshot_id IS NULL
),
source_processing_activity_vendors AS (
SELECT processing_activity_id, vendor_id, snapshot_id, created_at
FROM processing_activity_vendors
WHERE processing_activity_id = ANY(SELECT id FROM source_processing_activities)
),
source_vendors AS (
SELECT *
FROM vendors
WHERE %s AND id = ANY(SELECT vendor_id FROM source_processing_activity_vendors)
)
INSERT INTO vendors (
tenant_id,
id,
snapshot_id,
source_id,
organization_id,
name,
description,
category,
headquarter_address,
legal_name,
website_url,
privacy_policy_url,
service_level_agreement_url,
data_processing_agreement_url,
business_associate_agreement_url,
subprocessors_list_url,
certifications,
countries,
business_owner_profile_id,
security_owner_profile_id,
status_page_url,
terms_of_service_url,
security_page_url,
trust_page_url,
show_on_trust_center,
created_at,
updated_at
)
SELECT
@tenant_id,
generate_gid(decode_base64_unpadded(@tenant_id), @vendor_entity_type),
@snapshot_id,
v.id,
v.organization_id,
v.name,
v.description,
v.category,
v.headquarter_address,
v.legal_name,
v.website_url,
v.privacy_policy_url,
v.service_level_agreement_url,
v.data_processing_agreement_url,
v.business_associate_agreement_url,
v.subprocessors_list_url,
v.certifications,
v.countries,
v.business_owner_profile_id,
v.security_owner_profile_id,
v.status_page_url,
v.terms_of_service_url,
v.security_page_url,
v.trust_page_url,
v.show_on_trust_center,
v.created_at,
v.updated_at
FROM source_vendors v
`
query = fmt.Sprintf(query, scope.SQLFragment())
args := pgx.StrictNamedArgs{
"tenant_id": scope.GetTenantID(),
"snapshot_id": snapshotID,
"organization_id": organizationID,
"vendor_entity_type": VendorEntityType,
}
maps.Copy(args, scope.SQLArguments())
_, err := conn.Exec(ctx, query, args)
if err != nil {
return fmt.Errorf("cannot insert vendor snapshots for processing activities: %w", err)
}
return nil
}
func (v Vendors) Snapshot(ctx context.Context, conn pg.Tx, scope Scoper, organizationID, snapshotID gid.GID) error {
for _, snapshotter := range []VendorSnapshotter{
Vendors{},