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

Add assets snapshots

Browse Source 
Signed-off-by: Sacha Al Himdani <sacha@getprobo.com>
This commit is contained in:
Sacha Al Himdani
2025-08-29 19:05:17 +02:00
parent 612149afc8
commit 89036d3a0f
25 changed files with 839 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
pkg/coredata/asset.go
View File

@@ -29,6 +29,8 @@ import (
type (
Asset struct {
ID gid.GID `db:"id"`
SnapshotID *gid.GID `db:"snapshot_id"`
SourceID *gid.GID `db:"source_id"`
Name string `db:"name"`
Amount int `db:"amount"`
OwnerID gid.GID `db:"owner_id"`
@@ -65,6 +67,8 @@ func (a *Asset) LoadByID(
q := `
SELECT
id,
snapshot_id,
source_id,
name,
organization_id,
owner_id,
@@ -110,6 +114,8 @@ func (a *Asset) LoadByOwnerID(
q := `
SELECT
id,
snapshot_id,
source_id,
name,
organization_id,
owner_id,
@@ -152,6 +158,7 @@ func (a *Assets) CountByOrganizationID(
conn pg.Conn,
scope Scoper,
organizationID gid.GID,
filter *AssetFilter,
) (int, error) {
q := `
SELECT
@@ -161,12 +168,14 @@ FROM
WHERE
%s
AND organization_id = @organization_id
AND %s
`
q = fmt.Sprintf(q, scope.SQLFragment())
q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment())
args := pgx.StrictNamedArgs{"organization_id": organizationID}
maps.Copy(args, scope.SQLArguments())
maps.Copy(args, filter.SQLArguments())
row := conn.QueryRow(ctx, q, args)
@@ -184,10 +193,13 @@ func (a *Assets) LoadByOrganizationID(
scope Scoper,
organizationID gid.GID,
cursor *page.Cursor[AssetOrderField],
filter *AssetFilter,
) error {
q := `
SELECT
id,
snapshot_id,
source_id,
name,
organization_id,
owner_id,
@@ -203,12 +215,14 @@ WHERE
%s
AND organization_id = @organization_id
AND %s
AND %s
`
q = fmt.Sprintf(q, scope.SQLFragment(), cursor.SQLFragment())
q = fmt.Sprintf(q, scope.SQLFragment(), filter.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)
@@ -299,8 +313,11 @@ SET
WHERE
%s
AND id = @id
AND snapshot_id IS NULL
RETURNING
id,
snapshot_id,
source_id,
name,
organization_id,
owner_id,
@@ -351,6 +368,7 @@ DELETE FROM assets
WHERE
%s
AND id = @id
AND snapshot_id IS NULL
`
q = fmt.Sprintf(q, scope.SQLFragment())
@@ -365,3 +383,79 @@ WHERE
return nil
}
func (assets Assets) Snapshot(ctx context.Context, conn pg.Conn, scope Scoper, organizationID, snapshotID gid.GID) error {
snapshotters := []AssetSnapshotter{Assets{}, Vendors{}, AssetVendors{}}
for _, snapshotter := range snapshotters {
if err := snapshotter.InsertAssetSnapshots(ctx, conn, scope, organizationID, snapshotID); err != nil {
return fmt.Errorf("cannot create asset snapshots: (%T) %w", snapshotter, err)
}
}
return nil
}
func (assets Assets) InsertAssetSnapshots(
ctx context.Context,
conn pg.Conn,
scope Scoper,
organizationID gid.GID,
snapshotID gid.GID,
) error {
query := `
WITH
source_assets AS (
SELECT *
FROM assets
WHERE %s AND organization_id = @organization_id AND snapshot_id IS NULL
)
INSERT INTO assets (
tenant_id,
id,
snapshot_id,
source_id,
name,
organization_id,
owner_id,
amount,
criticity,
asset_type,
data_types_stored,
created_at,
updated_at
)
SELECT
@tenant_id,
generate_gid(decode_base64_unpadded(@tenant_id), @asset_entity_type),
@snapshot_id,
a.id,
a.name,
a.organization_id,
a.owner_id,
a.amount,
a.criticity,
a.asset_type,
a.data_types_stored,
a.created_at,
a.updated_at
FROM source_assets a
`
query = fmt.Sprintf(query, scope.SQLFragment())
args := pgx.StrictNamedArgs{
"tenant_id": scope.GetTenantID(),
"snapshot_id": snapshotID,
"organization_id": organizationID,
"asset_entity_type": AssetEntityType,
}
maps.Copy(args, scope.SQLArguments())
_, err := conn.Exec(ctx, query, args)
if err != nil {
return fmt.Errorf("cannot insert asset snapshots: %w", err)
}
return nil
}

54
pkg/coredata/asset_filter.go Normal file
View File

@@ -0,0 +1,54 @@
// Copyright (c) 2025 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/getprobo/probo/pkg/gid"
"github.com/jackc/pgx/v5"
)
type (
AssetFilter struct {
snapshotID **gid.GID
}
)
func NewAssetFilter(snapshotID **gid.GID) *AssetFilter {
return &AssetFilter{
snapshotID: snapshotID,
}
}
func (f *AssetFilter) SQLArguments() pgx.NamedArgs {
args := pgx.NamedArgs{}
if f.snapshotID != nil && *f.snapshotID != nil {
args["filter_snapshot_id"] = **f.snapshotID
}
return args
}
func (f *AssetFilter) SQLFragment() string {
if f.snapshotID == nil {
return "TRUE"
}
if *f.snapshotID == nil {
return "snapshot_id IS NULL"
} else {
return "snapshot_id = @filter_snapshot_id"
}
}

72
pkg/coredata/asset_vendor.go
View File

@@ -17,6 +17,7 @@ package coredata
import (
"context"
"fmt"
"maps"
"time"
"github.com/getprobo/probo/pkg/gid"
@@ -26,13 +27,18 @@ import (
type (
AssetVendor struct {
AssetID gid.GID `db:"asset_id"`
VendorID gid.GID `db:"vendor_id"`
TenantID gid.TenantID `db:"tenant_id"`
CreatedAt time.Time `db:"created_at"`
AssetID gid.GID `db:"asset_id"`
VendorID gid.GID `db:"vendor_id"`
TenantID gid.TenantID `db:"tenant_id"`
SnapshotID *gid.GID `db:"snapshot_id"`
CreatedAt time.Time `db:"created_at"`
}
AssetVendors []*AssetVendor
AssetSnapshotter interface {
InsertAssetSnapshots(ctx context.Context, conn pg.Conn, scope Scoper, organizationID, snapshotID gid.GID) error
}
)
func (av AssetVendors) Merge(
@@ -112,3 +118,61 @@ FROM vendor_ids
return nil
}
func (av AssetVendors) InsertAssetSnapshots(
ctx context.Context,
conn pg.Conn,
scope Scoper,
organizationID gid.GID,
snapshotID gid.GID,
) error {
query := `
WITH
source_assets AS (
SELECT id
FROM assets
WHERE organization_id = @organization_id AND snapshot_id IS NULL
),
snapshot_assets AS (
SELECT id, source_id
FROM assets
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_asset_vendors AS (
SELECT asset_id, vendor_id, snapshot_id, created_at
FROM asset_vendors
WHERE %s AND asset_id = ANY(SELECT id FROM source_assets) AND snapshot_id IS NULL
)
INSERT INTO asset_vendors (tenant_id, asset_id, vendor_id, snapshot_id, created_at)
SELECT
@tenant_id,
sa.id,
sv.id,
@snapshot_id,
av.created_at
FROM source_asset_vendors av
JOIN snapshot_assets sa ON sa.source_id = av.asset_id
JOIN snapshot_vendors sv ON sv.source_id = av.vendor_id
`
query = fmt.Sprintf(query, scope.SQLFragment())
args := pgx.StrictNamedArgs{
"tenant_id": scope.GetTenantID(),
"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 asset vendor snapshots: %w", err)
}
return nil
}

19
pkg/coredata/migrations/20250829T155558Z.sql Normal file
View File

@@ -0,0 +1,19 @@
ALTER TABLE assets ADD COLUMN snapshot_id TEXT;
ALTER TABLE assets ADD COLUMN source_id TEXT;
ALTER TABLE assets ADD CONSTRAINT assets_snapshot_id_fkey
FOREIGN KEY (snapshot_id)
REFERENCES snapshots(id)
ON UPDATE CASCADE
ON DELETE CASCADE;
ALTER TABLE assets ADD CONSTRAINT assets_source_id_snapshot_id_key
UNIQUE (source_id, snapshot_id);
ALTER TABLE asset_vendors ADD COLUMN snapshot_id TEXT;
ALTER TABLE asset_vendors ADD CONSTRAINT asset_vendors_snapshot_id_fkey
FOREIGN KEY (snapshot_id)
REFERENCES snapshots(id)
ON UPDATE CASCADE
ON DELETE CASCADE;

2
pkg/coredata/snapshottable.go
View File

@@ -28,6 +28,8 @@ type Snapshottable interface {
func GetSnapshottable(snapshotType SnapshotsType) (Snapshottable, error) {
switch snapshotType {
case SnapshotsTypeAssets:
return Assets{}, nil
case SnapshotsTypeData:
return Data{}, nil
case SnapshotsTypeNonConformityRegistries:

100
pkg/coredata/vendor.go
View File

@@ -814,3 +814,103 @@ FROM source_vendors v
return nil
}
func (vs Vendors) InsertAssetSnapshots(
ctx context.Context,
conn pg.Conn,
scope Scoper,
organizationID gid.GID,
snapshotID gid.GID,
) error {
query := `
WITH
source_assets AS (
SELECT id
FROM assets
WHERE organization_id = @organization_id AND snapshot_id IS NULL
),
source_asset_vendors AS (
SELECT asset_id, vendor_id, snapshot_id, created_at
FROM asset_vendors
WHERE asset_id = ANY(SELECT id FROM source_assets)
),
source_vendors AS (
SELECT *
FROM vendors
WHERE %s AND id = ANY(SELECT vendor_id FROM source_asset_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,
business_owner_id,
security_owner_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.business_owner_id,
v.security_owner_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 assets: %w", err)
}
return nil
}