From 71146ca0971841f93547a6a3dd95e119c9f01808 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Thu, 10 Sep 2026 18:11:31 +0200 Subject: [PATCH] feat: emit principal events on the firehose The firehose writes principal invalidation records as a second SSE event kind, `principal`, with tenant, principal, cause and recorded_at in DAG-JSON and the cause CID as the event id. api.FirehosePrincipalRevocation is the generated wire type; api.FirehoseEvent is a hand-written sum of the two firehose record types for consumers reading both kinds. GET /revocation/:cid is unaffected. The README documents the new event under GET /revocations/:from. Co-Authored-By: Claude Fable 5.1 --- README.md | 34 +++++--- pkg/api/gen/main.go | 2 +- pkg/api/json_gen.go | 189 ++++++++++++++++++++++++++++++++++++++++++ pkg/api/types.go | 44 ++++++++++ pkg/api/types_test.go | 61 ++++++++++++++ pkg/fx/app.go | 31 ++++++- pkg/fx/app_test.go | 127 +++++++++++++++++++++++++++- 7 files changed, 473 insertions(+), 15 deletions(-) create mode 100644 pkg/api/types_test.go diff --git a/README.md b/README.md index 7cccb81..12a8d8e 100644 --- a/README.md +++ b/README.md @@ -95,25 +95,39 @@ when no revocation exists. For example: "path": [ {"/": {"bytes": "omF2AWNjYXBsL3Rlc3QvaW52b2tl"}} ], - "recorded_at": "2026-07-17T09:00:00Z" + "recorded_at": 1784278800000000000 } ``` ### `GET /revocations/:from` -A Server-Sent Events stream of compact DAG-JSON records. Each event has `revoke` -(the revoked delegation CID), `path` (the witness delegation CIDs), `cause` (the -revocation invocation CID), and `recorded_at` (the time the record was recorded). Use `0` -to stream all stored records, or provide an RFC3339/RFC3339Nano timestamp cursor -to stream records recorded on or after it. The cursor is inclusive so consumers -resuming from the `recorded_at` of the last event they received do not miss -records that share it; deduplicate by the event `id` (the `cause` CID). For -example: +A Server-Sent Events stream of compact DAG-JSON records. Each `revocation` +event has `revoke` (the revoked delegation CID), `path` (the witness delegation +CIDs), `cause` (the revocation invocation CID), and `recorded_at` (the time the +record was recorded). Use `0` to stream all stored records, or provide an +RFC3339/RFC3339Nano timestamp cursor to stream records recorded on or after it. +The cursor is inclusive so consumers resuming from the `recorded_at` of the +last event they received do not miss records that share it; deduplicate by the +event `id` (the `cause` CID). For example: ```js id: bafyreif5fzax7oygfafacvxq2ndhtkshz2av5m42hqeixea7giirdxe5dm event: revocation -data: {"revoke":{"/":"bafyreiehytyi4q3t2amvf2abdlt5xnnqtaqkknf6yxhre4klpjnejlnsc4"},"path":[{"/":"bafyreiehytyi4q3t2amvf2abdlt5xnnqtaqkknf6yxhre4klpjnejlnsc4"}],"cause":{"/":"bafyreif5fzax7oygfafacvxq2ndhtkshz2av5m42hqeixea7giirdxe5dm"},"recorded_at":"2026-07-17T09:00:00Z"} +data: {"cause":{"/":"bafyreif5fzax7oygfafacvxq2ndhtkshz2av5m42hqeixea7giirdxe5dm"},"path":[{"/":"bafyreiehytyi4q3t2amvf2abdlt5xnnqtaqkknf6yxhre4klpjnejlnsc4"}],"recorded_at":1784278800000000000,"revoke":{"/":"bafyreiehytyi4q3t2amvf2abdlt5xnnqtaqkknf6yxhre4klpjnejlnsc4"}} +``` + +The same stream carries `principal` events, which record that every proof a +gateway cached for a principal's keys is void. Each has `tenant` (the tenant +DID), `principal` (the principal identifier, unique within the tenant), `cause` +(the invalidation invocation CID), and `recorded_at` (unix nanoseconds, as for every DAG-JSON time here). Keys are emitted in lexicographic order. The cursor, the inclusive +resume rule, and deduplication by `cause` are the same as for `revocation` +events. A principal revocation event revokes no delegation, so +`GET /revocation/:cid` never returns one. For example: + +```js +id: bafyreif5fzax7oygfafacvxq2ndhtkshz2av5m42hqeixea7giirdxe5dm +event: principal +data: {"cause":{"/":"bafyreif5fzax7oygfafacvxq2ndhtkshz2av5m42hqeixea7giirdxe5dm"},"principal":"8f2c","recorded_at":1788948000000000000,"tenant":"did:plc:tenant"} ``` ## Client library diff --git a/pkg/api/gen/main.go b/pkg/api/gen/main.go index 260ba32..f5d53ac 100644 --- a/pkg/api/gen/main.go +++ b/pkg/api/gen/main.go @@ -13,7 +13,7 @@ const buildTag = "//go:build !codegen\n\n" func main() { const output = "../json_gen.go" - if err := jsg.WriteMapEncodersToFile(output, "api", api.Revocation{}, api.FirehoseRevocation{}); err != nil { + if err := jsg.WriteMapEncodersToFile(output, "api", api.Revocation{}, api.FirehoseRevocation{}, api.FirehosePrincipalRevocation{}); err != nil { panic(err) } data, err := os.ReadFile(output) diff --git a/pkg/api/json_gen.go b/pkg/api/json_gen.go index 7531d65..c5e3951 100644 --- a/pkg/api/json_gen.go +++ b/pkg/api/json_gen.go @@ -525,3 +525,192 @@ func (t *FirehoseRevocation) UnmarshalDagJSON(r io.Reader) (err error) { return nil } +func (t *FirehosePrincipalRevocation) MarshalDagJSON(w io.Writer) error { + jw := jsg.NewDagJsonWriter(w) + if t == nil { + err := jw.WriteNull() + return err + } + if err := jw.WriteObjectOpen(); err != nil { + return err + } + written := false + + // t.Cause (cid.Cid) (struct) + if len("cause") > 8192 { + return fmt.Errorf("string in field \"cause\" was too long") + } + if err := jw.WriteString(string("cause")); err != nil { + return fmt.Errorf("writing string for field \"cause\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + + if err := jw.WriteCid(t.Cause); err != nil { + return fmt.Errorf("writing CID for field t.Cause: %w", err) + } + + written = true + if written { + if err := jw.WriteComma(); err != nil { + return err + } + } + + // t.Principal (string) (string) + if len("principal") > 8192 { + return fmt.Errorf("string in field \"principal\" was too long") + } + if err := jw.WriteString(string("principal")); err != nil { + return fmt.Errorf("writing string for field \"principal\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + if len(t.Principal) > 8192 { + return fmt.Errorf("string in field t.Principal was too long") + } + if err := jw.WriteString(string(t.Principal)); err != nil { + return fmt.Errorf("writing string for field t.Principal: %w", err) + } + written = true + if written { + if err := jw.WriteComma(); err != nil { + return err + } + } + + // t.RecordedAt (typegen.DagJsonTime) (struct) + if len("recorded_at") > 8192 { + return fmt.Errorf("string in field \"recorded_at\" was too long") + } + if err := jw.WriteString(string("recorded_at")); err != nil { + return fmt.Errorf("writing string for field \"recorded_at\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + if err := t.RecordedAt.MarshalDagJSON(jw); err != nil { + return fmt.Errorf("marshaling field t.RecordedAt: %w", err) + } + written = true + if written { + if err := jw.WriteComma(); err != nil { + return err + } + } + + // t.Tenant (did.DID) (struct) + if len("tenant") > 8192 { + return fmt.Errorf("string in field \"tenant\" was too long") + } + if err := jw.WriteString(string("tenant")); err != nil { + return fmt.Errorf("writing string for field \"tenant\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + if err := t.Tenant.MarshalDagJSON(jw); err != nil { + return fmt.Errorf("marshaling field t.Tenant: %w", err) + } + if err := jw.WriteObjectClose(); err != nil { + return err + } + return nil +} +func (t *FirehosePrincipalRevocation) UnmarshalDagJSON(r io.Reader) (err error) { + *t = FirehosePrincipalRevocation{} + + jr := jsg.NewDagJsonReader(r) + defer func() { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + }() + if err := jr.ReadObjectOpen(); err != nil { + return fmt.Errorf("reading object open for FirehosePrincipalRevocation: %w", err) + } + close, err := jr.PeekObjectClose() + if err != nil { + return fmt.Errorf("peeking object close for FirehosePrincipalRevocation: %w", err) + } + if close { + if err := jr.ReadObjectClose(); err != nil { + return fmt.Errorf("reading object close for FirehosePrincipalRevocation: %w", err) + } + } else { + for i := uint64(0); i < 8192; i++ { + name, err := jr.ReadString(8192) + if err != nil { + if errors.Is(err, jsg.ErrLimitExceeded) { + return fmt.Errorf("reading string for field FirehosePrincipalRevocation: string too large") + } + return fmt.Errorf("reading string for field FirehosePrincipalRevocation: %w", err) + } + if err := jr.ReadObjectColon(); err != nil { + return fmt.Errorf("reading object colon for field FirehosePrincipalRevocation: %w", err) + } + switch name { + + // t.Cause (cid.Cid) (struct) + case "cause": + { + + c, err := jr.ReadCid() + if err != nil { + return fmt.Errorf("reading CID for field t.Cause: %w", err) + } + t.Cause = c + + } + + // t.Principal (string) (string) + case "principal": + { + sval, err := jr.ReadString(8192) + if err != nil { + if errors.Is(err, jsg.ErrLimitExceeded) { + return fmt.Errorf("reading string for field t.Principal: string too long") + } + return fmt.Errorf("reading string for field t.Principal: %w", err) + } + t.Principal = string(sval) + } + + // t.RecordedAt (typegen.DagJsonTime) (struct) + case "recorded_at": + + if err := t.RecordedAt.UnmarshalDagJSON(jr); err != nil { + return fmt.Errorf("unmarshaling t.RecordedAt: %w", err) + } + + // t.Tenant (did.DID) (struct) + case "tenant": + + if err := t.Tenant.UnmarshalDagJSON(jr); err != nil { + return fmt.Errorf("unmarshaling t.Tenant: %w", err) + } + + default: + // Field doesn't exist on this type, so ignore it + if err := jr.DiscardType(); err != nil { + return fmt.Errorf("ignoring field %s for FirehosePrincipalRevocation: %w", name, err) + } + } + + close, err := jr.ReadObjectCloseOrComma() + if err != nil { + return fmt.Errorf("reading object close or comma for field FirehosePrincipalRevocation: %w", err) + } + if close { + break + } + if i == 8192-1 { + return fmt.Errorf("map too large for FirehosePrincipalRevocation") + } + } + } + + return nil +} diff --git a/pkg/api/types.go b/pkg/api/types.go index 6a2dc9a..e560cb8 100644 --- a/pkg/api/types.go +++ b/pkg/api/types.go @@ -3,6 +3,7 @@ package api import ( jsg "github.com/alanshaw/dag-json-gen" + "github.com/fil-forge/ucantone/did" "github.com/ipfs/go-cid" ) @@ -23,3 +24,46 @@ type FirehoseRevocation struct { Cause cid.Cid `dagjsongen:"cause"` RecordedAt jsg.DagJsonTime `dagjsongen:"recorded_at"` } + +// FirehosePrincipalRevocation is the representation of a principal +// invalidation emitted by the firehose. It names the principal whose cached +// proofs are void and no delegation. +type FirehosePrincipalRevocation struct { + Tenant did.DID `dagjsongen:"tenant"` + Principal string `dagjsongen:"principal"` + Cause cid.Cid `dagjsongen:"cause"` + RecordedAt jsg.DagJsonTime `dagjsongen:"recorded_at"` +} + +// FirehoseEvent is one event read from the firehose. Exactly one of +// Revocation and Principal is set, matching the SSE event name. +type FirehoseEvent struct { + Revocation *FirehoseRevocation + PrincipalRevocation *FirehosePrincipalRevocation +} + +// Cause returns the CID of the invocation that caused the event, or the +// undefined CID when the event carries no record. +func (e FirehoseEvent) Cause() cid.Cid { + switch { + case e.Revocation != nil: + return e.Revocation.Cause + case e.PrincipalRevocation != nil: + return e.PrincipalRevocation.Cause + default: + return cid.Undef + } +} + +// RecordedAt returns the time the event's record was recorded, or the zero +// time when the event carries no record. +func (e FirehoseEvent) RecordedAt() jsg.DagJsonTime { + switch { + case e.Revocation != nil: + return e.Revocation.RecordedAt + case e.PrincipalRevocation != nil: + return e.PrincipalRevocation.RecordedAt + default: + return jsg.DagJsonTime{} + } +} diff --git a/pkg/api/types_test.go b/pkg/api/types_test.go new file mode 100644 index 0000000..638b8f3 --- /dev/null +++ b/pkg/api/types_test.go @@ -0,0 +1,61 @@ +package api_test + +import ( + "bytes" + "testing" + "time" + + jsg "github.com/alanshaw/dag-json-gen" + "github.com/fil-forge/swarf/pkg/api" + "github.com/fil-forge/ucantone/did" + "github.com/ipfs/go-cid" + "github.com/stretchr/testify/require" +) + +func TestFirehosePrincipalRevocationRoundTrip(t *testing.T) { + tenant, err := did.Parse("did:plc:tenant") + require.NoError(t, err) + cause, err := cid.Decode("bafyreif5fzax7oygfafacvxq2ndhtkshz2av5m42hqeixea7giirdxe5dm") + require.NoError(t, err) + recordedAt := time.Date(2026, 9, 9, 10, 0, 0, 0, time.UTC) + value := api.FirehosePrincipalRevocation{ + Tenant: tenant, + Principal: "8f2c", + Cause: cause, + RecordedAt: jsg.DagJsonTime(recordedAt), + } + + var data bytes.Buffer + require.NoError(t, value.MarshalDagJSON(&data)) + require.JSONEq(t, `{ + "cause": {"/": "bafyreif5fzax7oygfafacvxq2ndhtkshz2av5m42hqeixea7giirdxe5dm"}, + "principal": "8f2c", + "recorded_at": 1788948000000000000, + "tenant": "did:plc:tenant" + }`, data.String()) + + var decoded api.FirehosePrincipalRevocation + require.NoError(t, decoded.UnmarshalDagJSON(&data)) + require.Equal(t, tenant, decoded.Tenant) + require.Equal(t, "8f2c", decoded.Principal) + require.Equal(t, cause, decoded.Cause) + require.True(t, decoded.RecordedAt.Time().Equal(recordedAt)) +} + +func TestFirehoseEvent(t *testing.T) { + cause, err := cid.Decode("bafyreif5fzax7oygfafacvxq2ndhtkshz2av5m42hqeixea7giirdxe5dm") + require.NoError(t, err) + recordedAt := jsg.DagJsonTime(time.Date(2026, 9, 9, 10, 0, 0, 0, time.UTC)) + + revocation := api.FirehoseEvent{Revocation: &api.FirehoseRevocation{Cause: cause, RecordedAt: recordedAt}} + require.Equal(t, cause, revocation.Cause()) + require.Equal(t, recordedAt, revocation.RecordedAt()) + + principal := api.FirehoseEvent{PrincipalRevocation: &api.FirehosePrincipalRevocation{Cause: cause, RecordedAt: recordedAt}} + require.Equal(t, cause, principal.Cause()) + require.Equal(t, recordedAt, principal.RecordedAt()) + + var empty api.FirehoseEvent + require.Equal(t, cid.Undef, empty.Cause()) + require.True(t, empty.RecordedAt().Time().IsZero()) +} diff --git a/pkg/fx/app.go b/pkg/fx/app.go index a619e15..44803db 100644 --- a/pkg/fx/app.go +++ b/pkg/fx/app.go @@ -7,6 +7,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "net" "net/http" "net/url" @@ -367,6 +368,8 @@ func writeFirehoseEvent(response *echo.Response, event store.Event) error { switch event.Kind { case store.EventKindRevocation: return writeFirehoseRecord(response, *event.Revocation) + case store.EventKindPrincipalRevocation: + return writeFirehosePrincipalRevocation(response, *event.PrincipalRevocation) default: return fmt.Errorf("unsupported firehose event kind %q", event.Kind) } @@ -377,7 +380,7 @@ func writeFirehoseRecord(response *echo.Response, record store.RevocationRecord) for i, delegation := range record.Path { path[i] = delegation.Link() } - data, err := encodeFirehoseEvent(api.FirehoseRevocation{ + data, err := encodeFirehoseEvent(&api.FirehoseRevocation{ Revoke: record.Revoke, Path: path, Cause: record.Cause.Link(), @@ -386,13 +389,30 @@ func writeFirehoseRecord(response *echo.Response, record store.RevocationRecord) if err != nil { return err } - if _, err := fmt.Fprintf(response, "id: %s\nevent: revocation\ndata: %s\n\n", record.Cause.Link(), data); err != nil { + if _, err := fmt.Fprintf(response, "id: %s\nevent: %s\ndata: %s\n\n", record.Cause.Link(), store.EventKindRevocation, data); err != nil { return fmt.Errorf("writing revocation event: %w", err) } response.Flush() return nil } +func writeFirehosePrincipalRevocation(response *echo.Response, record store.PrincipalRevocationRecord) error { + data, err := encodeFirehoseEvent(&api.FirehosePrincipalRevocation{ + Tenant: record.Tenant, + Principal: record.Principal, + Cause: record.Cause.Link(), + RecordedAt: jsg.DagJsonTime(record.RecordedAt), + }) + if err != nil { + return err + } + if _, err := fmt.Fprintf(response, "id: %s\nevent: %s\ndata: %s\n\n", record.Cause.Link(), store.EventKindPrincipalRevocation, data); err != nil { + return fmt.Errorf("writing principal event: %w", err) + } + response.Flush() + return nil +} + func encodeRevocationRecord(record store.RevocationRecord) ([]byte, error) { cause, err := invocation.Encode(record.Cause) if err != nil { @@ -419,7 +439,12 @@ func encodeRevocationRecord(record store.RevocationRecord) ([]byte, error) { return data.Bytes(), nil } -func encodeFirehoseEvent(event api.FirehoseRevocation) ([]byte, error) { +// dagJSONMarshaler is implemented by the generated firehose record types. +type dagJSONMarshaler interface { + MarshalDagJSON(w io.Writer) error +} + +func encodeFirehoseEvent(event dagJSONMarshaler) ([]byte, error) { var data bytes.Buffer if err := event.MarshalDagJSON(&data); err != nil { return nil, fmt.Errorf("encoding firehose record as DAG-JSON: %w", err) diff --git a/pkg/fx/app_test.go b/pkg/fx/app_test.go index 175f85c..d78d189 100644 --- a/pkg/fx/app_test.go +++ b/pkg/fx/app_test.go @@ -1,6 +1,7 @@ package fx import ( + "bufio" "context" "iter" "net/http" @@ -97,6 +98,118 @@ func TestFirehoseRouteStreamsRecords(t *testing.T) { require.NotContains(t, response.Body.String(), `"revocation"`) } +func TestFirehoseRouteStreamsPrincipalRevocationEvents(t *testing.T) { + id, err := identity.New("", "") + require.NoError(t, err) + command, err := command.Parse("/test/invalidate") + require.NoError(t, err) + revocation, err := invocation.Invoke(id, did.Undef, command, nil) + require.NoError(t, err) + witness, err := delegation.Delegate(id, did.Undef, id.DID(), command) + require.NoError(t, err) + invalidation, err := invocation.Invoke(id, did.Undef, command, nil, invocation.WithNonce([]byte("principal"))) + require.NoError(t, err) + tenant, err := did.Parse("did:plc:tenant") + require.NoError(t, err) + recordedAt := time.Now().UTC().Round(0) + source := &firehoseTestStore{events: []store.Event{ + store.RevocationEvent(store.RevocationRecord{Revoke: witness.Link(), Cause: revocation, Path: []ucan.Delegation{witness}, RecordedAt: recordedAt}), + store.PrincipalRevocationEvent(store.PrincipalRevocationRecord{Tenant: tenant, Principal: "alice", Cause: invalidation, RecordedAt: recordedAt.Add(time.Second)}), + }} + e := newEchoServer(id, server.NewHTTP(id), source) + request := httptest.NewRequest(http.MethodGet, "/revocations/0", nil) + response := httptest.NewRecorder() + e.ServeHTTP(response, request) + + require.Equal(t, http.StatusOK, response.Code) + frames := strings.Split(strings.TrimSuffix(response.Body.String(), "\n\n"), "\n\n") + require.Len(t, frames, 2) + + // Events are written in stream order, each under its own kind. + require.True(t, strings.HasPrefix(frames[0], "id: "+revocation.Link().String()+"\nevent: revocation\ndata: ")) + var revoked api.FirehoseRevocation + require.NoError(t, revoked.UnmarshalDagJSON(strings.NewReader(strings.TrimPrefix(frames[0], "id: "+revocation.Link().String()+"\nevent: revocation\ndata: ")))) + require.Equal(t, witness.Link(), revoked.Revoke) + + prefix := "id: " + invalidation.Link().String() + "\nevent: principal\ndata: " + require.True(t, strings.HasPrefix(frames[1], prefix), frames[1]) + var principal api.FirehosePrincipalRevocation + require.NoError(t, principal.UnmarshalDagJSON(strings.NewReader(strings.TrimPrefix(frames[1], prefix)))) + require.Equal(t, tenant, principal.Tenant) + require.Equal(t, "alice", principal.Principal) + require.Equal(t, invalidation.Link(), principal.Cause) + require.True(t, principal.RecordedAt.Time().Equal(recordedAt.Add(time.Second))) + require.NotContains(t, frames[1], `"revoke"`) + require.NotContains(t, frames[1], `"path"`) +} + +func TestFirehoseRouteStreamsMemoryStoreEvents(t *testing.T) { + id, err := identity.New("", "") + require.NoError(t, err) + command, err := command.Parse("/test/invalidate") + require.NoError(t, err) + revocation, err := invocation.Invoke(id, did.Undef, command, nil) + require.NoError(t, err) + witness, err := delegation.Delegate(id, did.Undef, id.DID(), command) + require.NoError(t, err) + invalidation, err := invocation.Invoke(id, did.Undef, command, nil, invocation.WithNonce([]byte("principal"))) + require.NoError(t, err) + tenant, err := did.Parse("did:plc:tenant") + require.NoError(t, err) + + records := memory.New() + require.NoError(t, records.Add(t.Context(), revocation, []ucan.Delegation{witness})) + require.NoError(t, records.AddPrincipalRevocation(t.Context(), invalidation, tenant, "alice")) + + srv := httptest.NewServer(newEchoServer(id, server.NewHTTP(id), records)) + defer srv.Close() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + request, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"/revocations/0", nil) + require.NoError(t, err) + response, err := http.DefaultClient.Do(request) + require.NoError(t, err) + defer response.Body.Close() + require.Equal(t, http.StatusOK, response.StatusCode) + require.Equal(t, "text/event-stream", response.Header.Get(echo.HeaderContentType)) + + // The stream stays open, so read exactly the two frames the store holds + // and then cancel the request. + scanner := bufio.NewScanner(response.Body) + var frames []map[string]string + frame := map[string]string{} + for len(frames) < 2 && scanner.Scan() { + line := scanner.Text() + if line == "" { + frames = append(frames, frame) + frame = map[string]string{} + continue + } + field, value, ok := strings.Cut(line, ": ") + require.True(t, ok, line) + frame[field] = value + } + require.NoError(t, scanner.Err()) + cancel() + require.Len(t, frames, 2) + + require.Equal(t, "revocation", frames[0]["event"]) + require.Equal(t, revocation.Link().String(), frames[0]["id"]) + var revoked api.FirehoseRevocation + require.NoError(t, revoked.UnmarshalDagJSON(strings.NewReader(frames[0]["data"]))) + require.Equal(t, witness.Link(), revoked.Revoke) + require.Equal(t, revocation.Link(), revoked.Cause) + + require.Equal(t, "principal", frames[1]["event"]) + require.Equal(t, invalidation.Link().String(), frames[1]["id"]) + var principal api.FirehosePrincipalRevocation + require.NoError(t, principal.UnmarshalDagJSON(strings.NewReader(frames[1]["data"]))) + require.Equal(t, tenant, principal.Tenant) + require.Equal(t, "alice", principal.Principal) + require.Equal(t, invalidation.Link(), principal.Cause) + require.False(t, principal.RecordedAt.Time().Before(revoked.RecordedAt.Time())) +} + func TestRevocationRouteReturnsDAGJSON(t *testing.T) { id, err := identity.New("", "") require.NoError(t, err) @@ -217,8 +330,12 @@ func TestValidateRevocationPath(t *testing.T) { require.Error(t, validateRevocationPath(t.Context(), []ucan.Delegation{expired}, resolver)) } +// firehoseTestStore is a fixed-content store. Get returns record; Stream +// yields events when set and otherwise record as a single revocation event, +// then ends. type firehoseTestStore struct { record store.RevocationRecord + events []store.Event from time.Time } @@ -237,6 +354,14 @@ func (s *firehoseTestStore) Get(context.Context, cid.Cid) (store.RevocationRecor func (s *firehoseTestStore) Stream(_ context.Context, from time.Time) iter.Seq2[store.Event, error] { s.from = from return func(yield func(store.Event, error) bool) { - yield(store.RevocationEvent(s.record), nil) + if s.events == nil { + yield(store.RevocationEvent(s.record), nil) + return + } + for _, event := range s.events { + if !yield(event, nil) { + return + } + } } }