From aef3832878d741d04eab8d4ff6f01794e5d8feb0 Mon Sep 17 00:00:00 2001 From: Paul Querna Date: Fri, 26 Jun 2026 21:59:50 +0000 Subject: [PATCH] feat(dotc1z): Exchanges() provenance sub-store (SDK side of object-request-provenance) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Companion to baton-axiomatic PR #175 (RFC 0001, per-object request provenance). Adds the SDK-side storage for captured transport exchanges so the export tool can resolve an emitted object's c1.connector.v2.RequestId annotations back to the (redacted) request/response that built it. - c1zstore.ExchangeStore + Exchange row type, reached via the OPTIONAL c1zstore.ExchangesProvider interface (type-asserted by the syncer) so engines and external Store implementations that predate provenance compile unchanged. - *C1File SQLite implementation: PutExchanges / GetExchange / ListExchangesForObject, with the exchanges table created on demand. Deliberately NOT registered in allTableDescriptors: exchanges are a per-sync debug side-store and must not enter the object sync-diff/clone machinery (which keys every table on external_id). Lazy creation also yields "old .c1z opens empty" for free. - Round-trip test + compile-time capability assertions. Not included (the cross-process handoff, RFC §6.7): the StreamCapturedExchanges ConnectorService RPC + the syncer-side consumer, which require a proto change in the c1 monorepo. This branch lands the storage shape so the RPC can target it. Co-authored-by: c1-squire-dev[bot] Co-Authored-By: Claude Opus 4.8 --- pkg/dotc1z/c1file_store.go | 10 +- pkg/dotc1z/c1zstore/exchanges.go | 46 +++++++++ pkg/dotc1z/exchanges.go | 172 +++++++++++++++++++++++++++++++ pkg/dotc1z/exchanges_test.go | 61 +++++++++++ 4 files changed, 285 insertions(+), 4 deletions(-) create mode 100644 pkg/dotc1z/c1zstore/exchanges.go create mode 100644 pkg/dotc1z/exchanges.go create mode 100644 pkg/dotc1z/exchanges_test.go diff --git a/pkg/dotc1z/c1file_store.go b/pkg/dotc1z/c1file_store.go index 1e6644ad0..921ec8af1 100644 --- a/pkg/dotc1z/c1file_store.go +++ b/pkg/dotc1z/c1file_store.go @@ -16,10 +16,12 @@ import ( // wrapper structs satisfy each sub-interface. These assertions catch // signature drift at build time rather than at the first runtime call. var ( - _ C1ZStore = (*C1File)(nil) - _ GrantStore = c1FileGrantStore{} - _ SyncMeta = c1FileSyncMeta{} - _ FileOps = c1FileFileOps{} + _ C1ZStore = (*C1File)(nil) + _ GrantStore = c1FileGrantStore{} + _ SyncMeta = c1FileSyncMeta{} + _ FileOps = c1FileFileOps{} + _ c1zstore.ExchangesProvider = (*C1File)(nil) + _ c1zstore.ExchangeStore = c1FileExchangeStore{} ) // Grants returns the grant-store slice of this c1z. diff --git a/pkg/dotc1z/c1zstore/exchanges.go b/pkg/dotc1z/c1zstore/exchanges.go new file mode 100644 index 000000000..93363e447 --- /dev/null +++ b/pkg/dotc1z/c1zstore/exchanges.go @@ -0,0 +1,46 @@ +package c1zstore + +import "context" + +// Exchange is one captured upstream transport request/response (HTTP/SQL/LDAP), +// keyed by RequestID, persisted alongside the synced objects so an operator can +// resolve an emitted object's c1.connector.v2.RequestId annotations back to the +// exchanges that built it. Request/Response are stored as already-redacted JSON +// (the connector redacts at capture). See the per-object-request-provenance RFC. +type Exchange struct { + RequestID string + SyncID string + TransportKind string // "http" | "sql" | "ldap" + // RequestJSON / ResponseJSON are redacted, transport-agnostic JSON blobs. + RequestJSON []byte + ResponseJSON []byte + // Intent routing context (surface/entrypoint/resourceType), JSON-encoded. + IntentJSON []byte + DiscoveredAt string +} + +// ExchangeStore is the provenance sub-store. It is reached via the optional +// ExchangesProvider interface (NOT embedded in Store) so that engines and +// external implementations that predate provenance keep compiling unchanged — +// the syncer type-asserts for ExchangesProvider and skips provenance when a +// store does not implement it. +type ExchangeStore interface { + // PutExchanges persists exchanges for the current sync (upsert by request id). + PutExchanges(ctx context.Context, exchanges ...*Exchange) error + // GetExchange returns the exchange for a request id, or (nil, nil) if absent. + GetExchange(ctx context.Context, requestID string) (*Exchange, error) + // ListExchangesForObject returns the exchanges for the given request ids + // (those an object's RequestId annotations point at), skipping any absent. + ListExchangesForObject(ctx context.Context, requestIDs []string) ([]*Exchange, error) +} + +// ExchangesProvider is the optional capability a Store implementation advertises +// when it can persist provenance exchanges. Callers do: +// +// if xp, ok := store.(c1zstore.ExchangesProvider); ok { xp.Exchanges()... } +// +// Absent → provenance is silently skipped; no behavior change, no break to +// external Store implementations. +type ExchangesProvider interface { + Exchanges() ExchangeStore +} diff --git a/pkg/dotc1z/exchanges.go b/pkg/dotc1z/exchanges.go new file mode 100644 index 000000000..d7fb230ad --- /dev/null +++ b/pkg/dotc1z/exchanges.go @@ -0,0 +1,172 @@ +package dotc1z + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/doug-martin/goqu/v9" + + "github.com/conductorone/baton-sdk/pkg/dotc1z/c1zstore" +) + +// ExchangeStore / Exchange re-exports so callers can use the dotc1z names, +// matching the convention used for GrantStore/SyncMeta/FileOps. +type ( + ExchangeStore = c1zstore.ExchangeStore + Exchange = c1zstore.Exchange +) + +const exchangesTableVersion = "1" +const exchangesTableName = "exchanges" +const exchangesTableSchema = ` +create table if not exists %s ( + id integer primary key, + request_id text not null, + sync_id text not null, + transport_kind text not null, + request_json blob, + response_json blob, + intent_json blob, + discovered_at datetime not null +); +create unique index if not exists %s on %s (request_id, sync_id);` + +var exchanges = (*exchangesTable)(nil) + +type exchangesTable struct{} + +func (r *exchangesTable) Name() string { return fmt.Sprintf("v%s_%s", r.Version(), exchangesTableName) } + +func (r *exchangesTable) Version() string { return exchangesTableVersion } + +func (r *exchangesTable) Schema() (string, []interface{}) { + return exchangesTableSchema, []interface{}{ + r.Name(), + fmt.Sprintf("idx_exchanges_request_sync_v%s", r.Version()), + r.Name(), + } +} + +func (r *exchangesTable) Migrations(ctx context.Context, db *goqu.Database) (bool, error) { + return false, nil +} + +// ensureExchangesTable creates the exchanges table on demand. The exchanges +// store is a per-sync debug side-store, deliberately NOT registered in +// allTableDescriptors: it must not participate in the object sync-diff / clone +// machinery (which keys every table on external_id). Creating it lazily also +// gives the "old .c1z opens empty" property for free — a file written before +// provenance simply never has the table, and reads treat its absence as empty. +func (c *C1File) ensureExchangesTable(ctx context.Context) error { + schema, args := exchanges.Schema() + q := fmt.Sprintf(schema, args...) + _, err := c.db.ExecContext(ctx, q) + return err +} + +// Exchanges returns the provenance exchange sub-store. Its presence is what +// makes *C1File satisfy c1zstore.ExchangesProvider. +func (c *C1File) Exchanges() ExchangeStore { return c1FileExchangeStore{c} } + +// c1FileExchangeStore is a zero-allocation value wrapper adapting *C1File to +// ExchangeStore. +type c1FileExchangeStore struct{ c *C1File } + +func (e c1FileExchangeStore) PutExchanges(ctx context.Context, items ...*Exchange) error { + return e.c.putExchanges(ctx, items...) +} + +func (e c1FileExchangeStore) GetExchange(ctx context.Context, requestID string) (*Exchange, error) { + return e.c.getExchange(ctx, requestID) +} + +func (e c1FileExchangeStore) ListExchangesForObject(ctx context.Context, requestIDs []string) ([]*Exchange, error) { + return e.c.listExchangesForObject(ctx, requestIDs) +} + +func (c *C1File) putExchanges(ctx context.Context, items ...*Exchange) error { + if c.readOnly { + return ErrReadOnly + } + if len(items) == 0 { + return nil + } + if err := c.validateSyncDb(ctx); err != nil { + return err + } + if err := c.ensureExchangesTable(ctx); err != nil { + return err + } + now := time.Now().Format("2006-01-02 15:04:05.999999999") + for _, it := range items { + if it == nil || it.RequestID == "" { + continue + } + fields := goqu.Record{ + "request_id": it.RequestID, + "sync_id": c.currentSyncID, + "transport_kind": it.TransportKind, + "request_json": it.RequestJSON, + "response_json": it.ResponseJSON, + "intent_json": it.IntentJSON, + "discovered_at": now, + } + q := c.db.Insert(exchanges.Name()).Prepared(true).Rows(fields) + q = q.OnConflict(goqu.DoUpdate("request_id, sync_id", + goqu.Record{ + "response_json": goqu.I("EXCLUDED.response_json"), + "request_json": goqu.I("EXCLUDED.request_json"), + })) + query, args, err := q.ToSQL() + if err != nil { + return err + } + if _, err := c.db.ExecContext(ctx, query, args...); err != nil { + return err + } + } + c.dbUpdated = true + return nil +} + +func (c *C1File) getExchange(ctx context.Context, requestID string) (*Exchange, error) { + if err := c.validateDb(ctx); err != nil { + return nil, err + } + q := c.db.From(exchanges.Name()).Prepared(true). + Select("request_id", "sync_id", "transport_kind", "request_json", "response_json", "intent_json", "discovered_at"). + Where(goqu.C("request_id").Eq(requestID)). + Order(goqu.C("id").Desc()). + Limit(1) + query, args, err := q.ToSQL() + if err != nil { + return nil, err + } + row := c.db.QueryRowContext(ctx, query, args...) + out := &Exchange{} + if err := row.Scan(&out.RequestID, &out.SyncID, &out.TransportKind, &out.RequestJSON, &out.ResponseJSON, &out.IntentJSON, &out.DiscoveredAt); err != nil { + // Absent rows, or an absent table on a .c1z written before provenance, + // both read as "no exchange". + if err.Error() == "sql: no rows in result set" || strings.Contains(err.Error(), "no such table") { + return nil, nil + } + return nil, err + } + return out, nil +} + +func (c *C1File) listExchangesForObject(ctx context.Context, requestIDs []string) ([]*Exchange, error) { + out := make([]*Exchange, 0, len(requestIDs)) + for _, id := range requestIDs { + ex, err := c.getExchange(ctx, id) + if err != nil { + return nil, err + } + if ex != nil { + out = append(out, ex) + } + } + return out, nil +} diff --git a/pkg/dotc1z/exchanges_test.go b/pkg/dotc1z/exchanges_test.go new file mode 100644 index 000000000..484c1a07e --- /dev/null +++ b/pkg/dotc1z/exchanges_test.go @@ -0,0 +1,61 @@ +package dotc1z + +import ( + "context" + "testing" + + "github.com/conductorone/baton-sdk/pkg/connectorstore" + "github.com/conductorone/baton-sdk/pkg/dotc1z/c1zstore" +) + +// Round-trips provenance exchanges through the SQLite Exchanges() sub-store and +// confirms the optional ExchangesProvider capability is advertised. This is the +// SDK side of the per-object-request-provenance handoff (RFC §6.6). +func TestExchangesSubStoreRoundTrip(t *testing.T) { + ctx := context.Background() + + f, err := NewC1ZFile(ctx, t.TempDir()+"/test.c1z") + if err != nil { + t.Fatalf("NewC1ZFile: %v", err) + } + defer f.Close(ctx) + + // Capability detection: the store advertises ExchangesProvider. + xp, ok := any(f).(c1zstore.ExchangesProvider) + if !ok { + t.Fatalf("C1File does not implement c1zstore.ExchangesProvider") + } + store := xp.Exchanges() + + if _, err := f.StartNewSync(ctx, connectorstore.SyncTypeFull, ""); err != nil { + t.Fatalf("StartNewSync: %v", err) + } + + in := []*c1zstore.Exchange{ + {RequestID: "req-s-1", TransportKind: "http", RequestJSON: []byte(`{"method":"GET","url":"/users"}`), ResponseJSON: []byte(`{"status":200}`)}, + {RequestID: "req-s-2", TransportKind: "http", RequestJSON: []byte(`{"method":"GET","url":"/users?cursor=c1"}`), ResponseJSON: []byte(`{"status":200}`)}, + } + if err := store.PutExchanges(ctx, in...); err != nil { + t.Fatalf("PutExchanges: %v", err) + } + + got, err := store.GetExchange(ctx, "req-s-1") + if err != nil || got == nil { + t.Fatalf("GetExchange(req-s-1) = %v, %v", got, err) + } + if got.TransportKind != "http" || string(got.ResponseJSON) != `{"status":200}` { + t.Fatalf("GetExchange round-trip mismatch: %+v", got) + } + + list, err := store.ListExchangesForObject(ctx, []string{"req-s-2", "req-missing", "req-s-1"}) + if err != nil { + t.Fatalf("ListExchangesForObject: %v", err) + } + if len(list) != 2 { + t.Fatalf("ListExchangesForObject returned %d, want 2 (missing id skipped)", len(list)) + } + + if missing, err := store.GetExchange(ctx, "nope"); err != nil || missing != nil { + t.Fatalf("GetExchange(absent) = %v, %v; want nil, nil", missing, err) + } +}