diff --git a/clickhouse/clickhouse.go b/clickhouse/clickhouse.go index d172798..e26b818 100644 --- a/clickhouse/clickhouse.go +++ b/clickhouse/clickhouse.go @@ -35,6 +35,10 @@ const ( IndexKeyColumn = "index_key" // DataIndexKeyColumn is the name of the data index name column in Clickhouse. DataIndexKeyColumn = "data_index_key" + // VoidsIDColumn is the name of the voids_id column in Clickhouse. For + // dimo.tombstone events it holds the id of the event, usually an attestation, + // being tombstoned; empty for all other event types. + VoidsIDColumn = "voids_id" // InsertStmt is the SQL statement for inserting a row into Clickhouse. InsertStmt = "INSERT INTO " + TableName + " (" + @@ -48,8 +52,9 @@ const ( DataVersionColumn + ", " + ExtrasColumn + ", " + IndexKeyColumn + ", " + - DataIndexKeyColumn + - ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" + DataIndexKeyColumn + ", " + + VoidsIDColumn + + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" // hexChars contains the characters used for hex representation hexChars = "0123456789abcdef" @@ -66,18 +71,18 @@ func CloudEventToSlice(event *cloudevent.CloudEventHeader) []any { // The order of the elements in the array match the order of the columns in the table. // This variant allows the caller to specify a value for index_key. func CloudEventToSliceWithKey(event *cloudevent.CloudEventHeader, key string) []any { - return cloudEventToSlice(event, key, "") + return cloudEventToSlice(event, key, "", "") } // StoredEventToSlice converts a StoredEvent to an array of any for Clickhouse -// insertion, populating both index_key (caller-supplied) and data_index_key -// (carried on the wrapper). The order of the elements matches the column -// order in the table. +// insertion, populating index_key (caller-supplied), data_index_key, and +// voids_id (both carried on the wrapper). The order of the elements matches +// the column order in the table. func StoredEventToSlice(stored *cloudevent.StoredEvent, indexKey string) []any { - return cloudEventToSlice(&stored.CloudEventHeader, indexKey, stored.DataIndexKey) + return cloudEventToSlice(&stored.CloudEventHeader, indexKey, stored.DataIndexKey, stored.VoidsID) } -func cloudEventToSlice(event *cloudevent.CloudEventHeader, indexKey, dataIndexKey string) []any { +func cloudEventToSlice(event *cloudevent.CloudEventHeader, indexKey, dataIndexKey, voidsID string) []any { // Add non-column fields to extras extras := cloudevent.AddNonColumnFieldsToExtras(event) @@ -99,6 +104,7 @@ func cloudEventToSlice(event *cloudevent.CloudEventHeader, indexKey, dataIndexKe string(jsonExtra), indexKey, dataIndexKey, + voidsID, } } @@ -108,11 +114,11 @@ func UnmarshalCloudEventSlice(jsonArray []byte) ([]any, error) { if err := json.Unmarshal(jsonArray, &rawSlice); err != nil { return nil, fmt.Errorf("failed to unmarshal cloud event slice: %w", err) } - if len(rawSlice) != 11 { + if len(rawSlice) != 12 { return nil, fmt.Errorf("invalid cloud event slice length: %d", len(rawSlice)) } - // Column order: subject, timestamp, eventType, id, source, producer, dataContentType, dataVersion, extras, indexKey, dataIndexKey + // Column order: subject, timestamp, eventType, id, source, producer, dataContentType, dataVersion, extras, indexKey, dataIndexKey, voidsID var ( subject string timestamp time.Time @@ -125,6 +131,7 @@ func UnmarshalCloudEventSlice(jsonArray []byte) ([]any, error) { extras string indexKey string dataIndexKey string + voidsID string ) unmarshal := func(i int, name string, ptr any) error { if err := json.Unmarshal(rawSlice[i], ptr); err != nil { @@ -165,7 +172,10 @@ func UnmarshalCloudEventSlice(jsonArray []byte) ([]any, error) { if err := unmarshal(10, "data index key", &dataIndexKey); err != nil { return nil, err } - return []any{subject, timestamp, eventType, id, source, producer, dataContentType, dataVersion, extras, indexKey, dataIndexKey}, nil + if err := unmarshal(11, "voids id", &voidsID); err != nil { + return nil, err + } + return []any{subject, timestamp, eventType, id, source, producer, dataContentType, dataVersion, extras, indexKey, dataIndexKey, voidsID}, nil } // CloudEventToObjectKey generates a unique key for storing cloud events. diff --git a/clickhouse/clickhouse_test.go b/clickhouse/clickhouse_test.go index fdecdf3..3f70f6f 100644 --- a/clickhouse/clickhouse_test.go +++ b/clickhouse/clickhouse_test.go @@ -33,7 +33,7 @@ func TestCloudEventToSlice(t *testing.T) { // Test CloudEventToSlice slice := CloudEventToSlice(event) - require.Len(t, slice, 11) + require.Len(t, slice, 12) // Verify the order and values of the slice assert.Equal(t, event.Subject, slice[0]) @@ -60,6 +60,9 @@ func TestCloudEventToSlice(t *testing.T) { // data_index_key is empty for CloudEventToSlice (payload is not external) assert.Equal(t, "", slice[10]) + + // voids_id is empty for non-tombstone events + assert.Equal(t, "", slice[11]) } func TestCloudEventToSliceWithKey(t *testing.T) { @@ -85,7 +88,7 @@ func TestCloudEventToSliceWithKey(t *testing.T) { customKey := "custom-key" slice := CloudEventToSliceWithKey(event, customKey) - require.Len(t, slice, 11) + require.Len(t, slice, 12) // Verify the order and values of the slice assert.Equal(t, event.Subject, slice[0]) @@ -111,6 +114,9 @@ func TestCloudEventToSliceWithKey(t *testing.T) { // data_index_key is empty for CloudEventToSliceWithKey (payload is not external) assert.Equal(t, "", slice[10]) + + // voids_id is empty for non-tombstone events + assert.Equal(t, "", slice[11]) } func TestStoredEventToSlice(t *testing.T) { @@ -135,11 +141,17 @@ func TestStoredEventToSlice(t *testing.T) { } slice := StoredEventToSlice(stored, "bundle/key#7") - require.Len(t, slice, 11) + require.Len(t, slice, 12) assert.Equal(t, stored.Subject, slice[0]) assert.Equal(t, "bundle/key#7", slice[9]) assert.Equal(t, "payloads/abc123.jpg", slice[10]) + assert.Equal(t, "", slice[11]) + + // VoidsID on the wrapper propagates into the slice. + stored.VoidsID = "voided-id-1" + slice = StoredEventToSlice(stored, "bundle/key#7") + assert.Equal(t, "voided-id-1", slice[11]) } func TestUnmarshalCloudEventSlice(t *testing.T) { @@ -158,6 +170,7 @@ func TestUnmarshalCloudEventSlice(t *testing.T) { `{"extra1":"value1","extra2":123}`, "test-key", "test-data-key", + "test-voids-id", } // Marshal the slice to JSON diff --git a/clickhouse/migrations/00008_add_voids_id_migration.sql b/clickhouse/migrations/00008_add_voids_id_migration.sql new file mode 100644 index 0000000..7719ec9 --- /dev/null +++ b/clickhouse/migrations/00008_add_voids_id_migration.sql @@ -0,0 +1,20 @@ +-- +goose Up +-- +goose StatementBegin +ALTER TABLE cloud_event ADD COLUMN voids_id String DEFAULT '' COMMENT 'For dimo.tombstone events, the id of the event being tombstoned. Empty for all other event types.' AFTER data_index_key; +-- +goose StatementEnd +-- +goose StatementBegin +-- Make it cheap to find these events. This may also help with some attestation filtering. +ALTER TABLE cloud_event ADD INDEX idx_event_type event_type TYPE set(0) GRANULARITY 1; +-- +goose StatementEnd +-- +goose StatementBegin +-- Do this asynchronously. +ALTER TABLE cloud_event MATERIALIZE INDEX idx_event_type; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +ALTER TABLE cloud_event DROP INDEX idx_event_type; +-- +goose StatementEnd +-- +goose StatementBegin +ALTER TABLE cloud_event DROP COLUMN voids_id; +-- +goose StatementEnd diff --git a/clickhouse/migrations/migrations_test.go b/clickhouse/migrations/migrations_test.go index d027e58..c889a92 100644 --- a/clickhouse/migrations/migrations_test.go +++ b/clickhouse/migrations/migrations_test.go @@ -60,6 +60,7 @@ func TestMigration(t *testing.T) { localch.ExtrasColumn, localch.IndexKeyColumn, localch.DataIndexKeyColumn, + localch.VoidsIDColumn, } assert.ElementsMatch(t, expectedCols, cols, "Columns do not match") diff --git a/cloudevent_types.go b/cloudevent_types.go index dae6d9c..0ee101c 100644 --- a/cloudevent_types.go +++ b/cloudevent_types.go @@ -17,6 +17,12 @@ const ( // TypeAttestation is the event type for 3rd party attestations TypeAttestation = "dimo.attestation" + // TypeAttestationTombstone is the event type used to tombstone an + // attestation. The data payload carries the id of the attestation + // being tombstoned in the voidsId field; see the voids_id ClickHouse + // column populated from it. + TypeAttestationTombstone = "dimo.tombstone" + // TypeUnknown is the event type for unknown events. TypeUnknown = "dimo.unknown" diff --git a/stored_event.go b/stored_event.go index a0e8506..b42fb8c 100644 --- a/stored_event.go +++ b/stored_event.go @@ -8,11 +8,20 @@ package cloudevent // an external object holding the payload. In that case the embedded // RawEvent's Data and DataBase64 fields will typically be empty. // +// VoidsID, when non-empty, is the id of the attestation that this event +// tombstones. It is only meaningful for events whose Type is +// TypeAttestationTombstone and is extracted server-side from the +// signed Data payload — it must never be set from producer-supplied input +// directly. It is used to cheaply populate the voids_id column in ClickHouse, +// but is not stored as a column in Parquet bundles: you would have to parse +// the data section of a Parquet row to recover the id. +// // StoredEvent is deliberately separate from CloudEventHeader / RawEvent so // that the wire-format types remain pure and shared safely across services -// — DataIndexKey points into trusted internal storage and must never be set -// from producer-supplied input. +// — DataIndexKey and VoidsID point into trusted internal storage and must +// never be set from producer-supplied input. type StoredEvent struct { RawEvent DataIndexKey string + VoidsID string }