feat: emit principal events on the firehose

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

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)

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{}
+	}
+}

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())
+}