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

Data as document: replace snapshot with publish workflow

Browse Source 
Mirror the SOA-to-document migration for the data list. Remove data
from the snapshot system and add a publish workflow that generates a
ProseMirror document for the full organization data inventory.

Signed-off-by: Sacha Al Himdani <sacha@getprobo.com>
This commit is contained in:
Sacha Al Himdani
2026-04-16 13:50:30 +02:00
parent 472ca703b5
commit 55563226cb
37 changed files with 2021 additions and 467 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
pkg/coredata/datum.go
View File

@@ -34,17 +34,11 @@ type (
OrganizationID gid.GID `db:"organization_id"`
OwnerID gid.GID `db:"owner_profile_id"`
DataClassification DataClassification `db:"data_classification"`
SnapshotID *gid.GID `db:"snapshot_id"`
SourceID *gid.GID `db:"source_id"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
Data []*Datum
DataSnapshotter interface {
InsertDataSnapshots(ctx context.Context, conn pg.Tx, scope Scoper, organizationID, snapshotID gid.GID) error
}
)
func (d *Datum) CursorKey(field DatumOrderField) page.CursorKey {
@@ -88,8 +82,6 @@ SELECT
owner_profile_id,
organization_id,
data_classification,
snapshot_id,
source_id,
created_at,
updated_at
FROM
@@ -97,6 +89,7 @@ FROM
WHERE
%s
AND id = @data_id
AND snapshot_id IS NULL
LIMIT 1;
`
@@ -132,8 +125,6 @@ SELECT
owner_profile_id,
organization_id,
data_classification,
snapshot_id,
source_id,
created_at,
updated_at
FROM
@@ -141,6 +132,7 @@ FROM
WHERE
%s
AND owner_profile_id = @owner_profile_id
AND snapshot_id IS NULL
LIMIT 1;
`
@@ -169,7 +161,6 @@ func (d *Data) CountByOrganizationID(
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
filter *DatumFilter,
) (int, error) {
q := `
SELECT
@@ -179,14 +170,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)
@@ -205,7 +195,6 @@ func (d *Data) LoadByOrganizationID(
scope Scoper,
organizationID gid.GID,
cursor *page.Cursor[DatumOrderField],
filter *DatumFilter,
) error {
q := `
SELECT
@@ -214,8 +203,6 @@ SELECT
organization_id,
owner_profile_id,
data_classification,
snapshot_id,
source_id,
created_at,
updated_at
FROM
@@ -223,15 +210,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)
@@ -249,6 +235,51 @@ WHERE
return nil
}
func (d *Data) LoadAllByOrganizationID(
ctx context.Context,
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
) error {
q := `
SELECT
id,
name,
organization_id,
owner_profile_id,
data_classification,
created_at,
updated_at
FROM
data
WHERE
%s
AND organization_id = @organization_id
AND snapshot_id IS NULL
ORDER BY
name ASC
`
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{"organization_id": organizationID}
maps.Copy(args, scope.SQLArguments())
rows, err := conn.Query(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot query data: %w", err)
}
data, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[Datum])
if err != nil {
return fmt.Errorf("cannot collect data: %w", err)
}
*d = data
return nil
}
func (d *Datum) Insert(
ctx context.Context,
conn pg.Tx,
@@ -262,8 +293,6 @@ INSERT INTO data (
owner_profile_id,
organization_id,
data_classification,
snapshot_id,
source_id,
created_at,
updated_at
) VALUES (
@@ -273,8 +302,6 @@ INSERT INTO data (
@owner_profile_id,
@organization_id,
@data_classification,
@snapshot_id,
@source_id,
@created_at,
@updated_at
)
@@ -287,8 +314,6 @@ INSERT INTO data (
"owner_profile_id": d.OwnerID,
"organization_id": d.OrganizationID,
"data_classification": d.DataClassification,
"snapshot_id": d.SnapshotID,
"source_id": d.SourceID,
"created_at": d.CreatedAt,
"updated_at": d.UpdatedAt,
}
@@ -323,8 +348,6 @@ RETURNING
owner_profile_id,
organization_id,
data_classification,
snapshot_id,
source_id,
created_at,
updated_at
`
@@ -380,73 +403,3 @@ WHERE
return nil
}
func (d Data) Snapshot(ctx context.Context, conn pg.Tx, scope Scoper, organizationID, snapshotID gid.GID) error {
snapshotters := []DataSnapshotter{Data{}, Vendors{}, DatumVendors{}}
for _, snapshotter := range snapshotters {
if err := snapshotter.InsertDataSnapshots(ctx, conn, scope, organizationID, snapshotID); err != nil {
return fmt.Errorf("cannot create data snapshots: (%T) %w", snapshotter, err)
}
}
return nil
}
func (d Data) InsertDataSnapshots(
ctx context.Context,
conn pg.Tx,
scope Scoper,
organizationID gid.GID,
snapshotID gid.GID,
) error {
query := `
WITH
source_data AS (
SELECT *
FROM data
WHERE %s AND organization_id = @organization_id AND snapshot_id IS NULL
)
INSERT INTO data (
tenant_id,
id,
snapshot_id,
source_id,
name,
organization_id,
owner_profile_id,
data_classification,
created_at,
updated_at
)
SELECT
@tenant_id,
generate_gid(decode_base64_unpadded(@tenant_id), @datum_entity_type),
@snapshot_id,
d.id,
d.name,
d.organization_id,
d.owner_profile_id,
d.data_classification,
d.created_at,
d.updated_at
FROM source_data d
`
query = fmt.Sprintf(query, scope.SQLFragment())
args := pgx.StrictNamedArgs{
"tenant_id": scope.GetTenantID(),
"snapshot_id": snapshotID,
"organization_id": organizationID,
"datum_entity_type": DatumEntityType,
}
maps.Copy(args, scope.SQLArguments())
_, err := conn.Exec(ctx, query, args)
if err != nil {
return fmt.Errorf("cannot insert data snapshots: %w", err)
}
return nil
}

54
pkg/coredata/datum_filter.go
View File

@@ -1,54 +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 (
DatumFilter struct {
snapshotID **gid.GID
}
)
func NewDatumFilter(snapshotID **gid.GID) *DatumFilter {
return &DatumFilter{
snapshotID: snapshotID,
}
}
func (f *DatumFilter) SQLArguments() pgx.NamedArgs {
args := pgx.NamedArgs{}
if f.snapshotID != nil && *f.snapshotID != nil {
args["filter_snapshot_id"] = **f.snapshotID
}
return args
}
func (f *DatumFilter) SQLFragment() string {
if f.snapshotID == nil {
return "TRUE"
}
if *f.snapshotID == nil {
return "snapshot_id IS NULL"
} else {
return "snapshot_id = @filter_snapshot_id"
}
}

66
pkg/coredata/datum_vendor.go
View File

@@ -17,7 +17,6 @@ package coredata
import (
"context"
"fmt"
"maps"
"time"
"github.com/jackc/pgx/v5"
@@ -27,10 +26,9 @@ import (
type (
DatumVendor struct {
DatumID gid.GID `db:"datum_id"`
VendorID gid.GID `db:"vendor_id"`
SnapshotID *gid.GID `db:"snapshot_id"`
CreatedAt time.Time `db:"created_at"`
DatumID gid.GID `db:"datum_id"`
VendorID gid.GID `db:"vendor_id"`
CreatedAt time.Time `db:"created_at"`
}
DatumVendors []*DatumVendor
@@ -119,61 +117,3 @@ FROM vendor_ids
return nil
}
func (d DatumVendors) InsertDataSnapshots(
ctx context.Context,
conn pg.Tx,
scope Scoper,
organizationID gid.GID,
snapshotID gid.GID,
) error {
query := `
WITH
source_data AS (
SELECT id
FROM data
WHERE organization_id = @organization_id AND snapshot_id IS NULL
),
snapshot_data AS (
SELECT id, source_id
FROM data
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_data_vendors AS (
SELECT datum_id, vendor_id, snapshot_id, created_at
FROM data_vendors
WHERE %s AND datum_id = ANY(SELECT id FROM source_data)
)
INSERT INTO data_vendors (tenant_id, datum_id, vendor_id, organization_id, snapshot_id, created_at)
SELECT
@tenant_id,
sd.id,
sv.id,
@organization_id,
@snapshot_id,
dv.created_at
FROM source_data_vendors dv
JOIN snapshot_data sd ON sd.source_id = dv.datum_id
JOIN snapshot_vendors sv ON sv.source_id = dv.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 datum vendor snapshots: %w", err)
}
return nil
}

22
pkg/coredata/migrations/20260420T120000Z.sql Normal file
View File

@@ -0,0 +1,22 @@
-- 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.
CREATE TABLE generated_documents (
organization_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE,
tenant_id TEXT NOT NULL,
data_document_id TEXT REFERENCES documents(id) ON DELETE SET NULL,
created_at TIMESTAMP WITH TIME ZONE NOT NULL,
updated_at TIMESTAMP WITH TIME ZONE NOT NULL,
PRIMARY KEY (organization_id)
);

4
pkg/coredata/snapshot.go
View File

@@ -124,7 +124,7 @@ FROM
WHERE
%s
AND organization_id = @organization_id
AND type != 'STATEMENTS_OF_APPLICABILITY'
AND type NOT IN ('STATEMENTS_OF_APPLICABILITY', 'DATA')
AND %s
`
@@ -164,7 +164,7 @@ FROM
WHERE
%s
AND organization_id = @organization_id
AND type != 'STATEMENTS_OF_APPLICABILITY'
AND type NOT IN ('STATEMENTS_OF_APPLICABILITY', 'DATA')
AND %s
`

1
pkg/coredata/snapshots_type.go
View File

@@ -39,7 +39,6 @@ func SnapshotsTypes() []SnapshotsType {
SnapshotsTypeRisks,
SnapshotsTypeVendors,
SnapshotsTypeAssets,
SnapshotsTypeData,
SnapshotsTypeFindings,
SnapshotsTypeObligations,
SnapshotsTypeProcessingActivities,

2
pkg/coredata/snapshottable.go
View File

@@ -32,8 +32,6 @@ func GetSnapshottable(snapshotType SnapshotsType) (Snapshottable, error) {
return Assets{}, nil
case SnapshotsTypeRisks:
return Risks{}, nil
case SnapshotsTypeData:
return Data{}, nil
case SnapshotsTypeFindings:
return Findings{}, nil
case SnapshotsTypeObligations:

198
pkg/coredata/vendor.go
View File

@@ -717,6 +717,102 @@ WHERE %s
return count, nil
}
func (vs *Vendors) LoadAllByDatumID(
ctx context.Context,
conn pg.Querier,
scope Scoper,
datumID gid.GID,
) error {
q := `
WITH vend AS (
SELECT
v.id,
v.tenant_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.snapshot_id,
v.source_id,
v.created_at,
v.updated_at
FROM
vendors v
INNER JOIN
data_vendors dv ON v.id = dv.vendor_id
WHERE
dv.datum_id = @datum_id
)
SELECT
id,
tenant_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,
snapshot_id,
source_id,
created_at,
updated_at
FROM
vend
WHERE %s
ORDER BY name ASC
`
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{"datum_id": datumID}
maps.Copy(args, scope.SQLArguments())
rows, err := conn.Query(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot query vendors: %w", err)
}
vendors, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[Vendor])
if err != nil {
return fmt.Errorf("cannot collect vendors: %w", err)
}
*vs = vendors
return nil
}
func (vs *Vendors) LoadByDatumID(
ctx context.Context,
conn pg.Querier,
@@ -981,108 +1077,6 @@ ORDER BY
return vendorMap, nil
}
func (d Vendors) InsertDataSnapshots(
ctx context.Context,
conn pg.Tx,
scope Scoper,
organizationID gid.GID,
snapshotID gid.GID,
) error {
query := `
WITH
source_data AS (
SELECT id
FROM data
WHERE organization_id = @organization_id AND snapshot_id IS NULL
),
source_data_vendors AS (
SELECT datum_id, vendor_id, snapshot_id, created_at
FROM data_vendors
WHERE datum_id = ANY(SELECT id FROM source_data)
),
source_vendors AS (
SELECT *
FROM vendors
WHERE %s AND id = ANY(SELECT vendor_id FROM source_data_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: %w", err)
}
return nil
}
func (vs Vendors) InsertAssetSnapshots(
ctx context.Context,
conn pg.Tx,