add plc event metadata

.github/workflows/docker.yml

···
         uses: docker/setup-buildx-action@v3
 
       - name: Log in to GitHub Container Registry
-        if: github.event_name != 'pull_request'
+        if: github.event_name != 'pull_request'amd
         uses: docker/login-action@v3
         with:
           registry: ${{ env.REGISTRY }}
···
           labels: ${{ steps.meta.outputs.labels }}
           cache-from: type=gha
           cache-to: type=gha,mode=max
-          platforms: linux/amd64,linux/arm64
+          platforms: linux/amd64

atkafka/atkafka.go

···
 	"os"
 	"os/signal"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 
 	"github.com/bluesky-social/indigo/atproto/atdata"
+	"github.com/bluesky-social/indigo/atproto/identity"
 	"github.com/bluesky-social/indigo/events"
 	"github.com/bluesky-social/indigo/events/schedulers/parallel"
 	"github.com/bluesky-social/indigo/repo"
···
 
 	atkProducer *Producer
 	ospProducer *Producer
+
+	plcClient *PlcClient
 }
 
 type ServerArgs struct {
 	RelayHost        string
+	PlcHost          string
 	BootstrapServers []string
 	OutputTopic      string
 	OspreyCompat     bool
···
 		args.Logger = slog.Default()
 	}
 
+	var plcClient *PlcClient
+	if args.PlcHost != "" {
+		plcClient = NewPlcClient(&PlcClientArgs{
+			PlcHost: args.PlcHost,
+		})
+	}
+
 	return &Server{
 		relayHost:        args.RelayHost,
+		plcClient:        plcClient,
 		bootstrapServers: args.BootstrapServers,
 		outputTopic:      args.OutputTopic,
 		ospreyCompat:     args.OspreyCompat,
···
 	return nil
 }
 
+type EventMetadata struct {
+	DidDocument  *identity.DIDDocument `json:"didDocument,omitempty"`
+	PdsHost      string                `json:"pdsHost,omitempty"`
+	DidCreatedAt string                `json:"didCreatedAt,omitempty"`
+	AccountAge   int64                 `json:"accountAge"`
+}
+
+func (s *Server) FetchEventMetadata(ctx context.Context, did string) (*EventMetadata, error) {
+	var didDocument *identity.DIDDocument
+	var pdsHost string
+	var didCreatedAt string
+	accountAge := int64(-1)
+
+	var wg sync.WaitGroup
+
+	if s.plcClient != nil {
+		wg.Go(func() {
+			logger := s.logger.With("component", "didDoc")
+			doc, err := s.plcClient.GetDIDDoc(ctx, did)
+			if err != nil {
+				logger.Error("error fetching did doc", "did", did, "err", err)
+				return
+			}
+			didDocument = doc
+
+			for _, svc := range doc.Service {
+				if svc.ID == "#atproto_pds" {
+					pdsHost = svc.ServiceEndpoint
+					break
+				}
+			}
+		})
+
+		wg.Go(func() {
+			logger := s.logger.With("component", "auditLog")
+			auditLog, err := s.plcClient.GetDIDAuditLog(ctx, did)
+			if err != nil {
+				logger.Error("error fetching did audit log", "did", did, "err", err)
+				return
+			}
+
+			didCreatedAt = auditLog.CreatedAt
+
+			createdAt, err := time.Parse(time.RFC3339Nano, auditLog.CreatedAt)
+			if err != nil {
+				logger.Error("error parsing timestamp in audit log", "did", did, "timestamp", auditLog.CreatedAt, "err", err)
+				return
+			}
+
+			accountAge = int64(time.Since(createdAt).Seconds())
+		})
+	}
+
+	wg.Wait()
+
+	return &EventMetadata{
+		DidDocument:  didDocument,
+		PdsHost:      pdsHost,
+		DidCreatedAt: didCreatedAt,
+		AccountAge:   accountAge,
+	}, nil
+}
+
 func (s *Server) handleEvent(ctx context.Context, evt *events.XRPCStreamEvent) error {
+	dispatchCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+
 	logger := s.logger.With("component", "handleEvent")
 	logger.Debug("event", "seq", evt.Sequence())
 
···
 				Operation: &atkOp,
 			}
 
+			eventMetadata, err := s.FetchEventMetadata(dispatchCtx, evt.RepoCommit.Repo)
+			if err != nil {
+				logger.Error("error fetching event metadata", "err", err)
+			} else {
+				kafkaEvt.Metadata = eventMetadata
+			}
+
 			var kafkaEvtBytes []byte
 			if s.ospreyCompat {
-
 				// create the wrapper event for osprey
 				ospreyKafkaEvent := OspreyAtKafkaEvent{
 					Data: OspreyEventData{
···
 			return fmt.Errorf("unhandled event received")
 		}
 
+		if did != "" {
+			eventMetadata, err := s.FetchEventMetadata(dispatchCtx, did)
+			if err != nil {
+				logger.Error("error fetching event metadata", "err", err)
+			} else {
+				kafkaEvt.Metadata = eventMetadata
+			}
+		}
+
 		// create the kafka event bytes
 		var kafkaEvtBytes []byte
 		var err error
···
 }
 
 type AtKafkaEvent struct {
-	Did       string `json:"did"`
-	Timestamp string `json:"timestamp"`
+	Did       string         `json:"did"`
+	Timestamp string         `json:"timestamp"`
+	Metadata  *EventMetadata `json:"eventMetadata"`
 
 	Operation *AtKafkaOp       `json:"operation,omitempty"`
 	Account   *AtKafkaAccount  `json:"account,omitempty"`

atkafka/plc.go

···
+package atkafka
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"time"
+
+	"github.com/bluesky-social/indigo/atproto/identity"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+	"github.com/bluesky-social/indigo/pkg/robusthttp"
+	lru "github.com/hashicorp/golang-lru/v2/expirable"
+	"golang.org/x/time/rate"
+)
+
+type PlcClient struct {
+	client     *http.Client
+	dir        *identity.BaseDirectory
+	plcHost    string
+	docCache   *lru.LRU[string, *identity.DIDDocument]
+	auditCache *lru.LRU[string, *DidAuditEntry]
+}
+
+type PlcClientArgs struct {
+	PlcHost string
+}
+
+func NewPlcClient(args *PlcClientArgs) *PlcClient {
+	client := robusthttp.NewClient(robusthttp.WithMaxRetries(2))
+	client.Timeout = 3 * time.Second
+
+	baseDir := identity.BaseDirectory{
+		PLCURL:     args.PlcHost,
+		PLCLimiter: rate.NewLimiter(rate.Limit(200), 100),
+		HTTPClient: *client,
+		Resolver: net.Resolver{
+			PreferGo: true,
+			Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
+				dialer := net.Dialer{Timeout: time.Second * 5}
+				nameserver := address
+				return dialer.DialContext(ctx, network, nameserver)
+			},
+		},
+		TryAuthoritativeDNS: true,
+		// primary Bluesky PDS instance only supports HTTP resolution method
+		SkipDNSDomainSuffixes: []string{".bsky.social"},
+	}
+
+	docCache := lru.NewLRU[string, *identity.DIDDocument](100_000, nil, 5*time.Minute)
+
+	auditCache := lru.NewLRU[string, *DidAuditEntry](100_000, nil, 1*time.Hour)
+
+	return &PlcClient{
+		client:     client,
+		dir:        &baseDir,
+		plcHost:    args.PlcHost,
+		docCache:   docCache,
+		auditCache: auditCache,
+	}
+}
+
+type OperationService struct {
+	Type     string `json:"type"`
+	Endpoint string `json:"endpoint"`
+}
+
+type DidLogEntry struct {
+	Sig                 string                      `json:"sig"`
+	Prev                *string                     `json:"prev"`
+	Type                string                      `json:"type"`
+	Services            map[string]OperationService `json:"services"`
+	AlsoKnownAs         []string                    `json:"alsoKnownAs"`
+	RotationKeys        []string                    `json:"rotationKeys"`
+	VerificationMethods map[string]string           `json:"verificationMethods"`
+}
+
+type DidAuditEntry struct {
+	Did       string      `json:"did"`
+	Operation DidLogEntry `json:"operation"`
+	Cid       string      `json:"cid"`
+	Nullified bool        `json:"nullified"`
+	CreatedAt string      `json:"createdAt"`
+}
+
+type DidAuditLog []DidAuditEntry
+
+func (c *PlcClient) GetDIDDoc(ctx context.Context, did string) (*identity.DIDDocument, error) {
+	if val, ok := c.docCache.Get(did); ok {
+		return val, nil
+	}
+
+	didDoc, err := c.dir.ResolveDID(ctx, syntax.DID(did))
+	if err != nil {
+		return nil, fmt.Errorf("failed to lookup DID: %w", err)
+	}
+
+	if didDoc == nil {
+		return nil, fmt.Errorf("DID Document not found")
+	}
+
+	c.docCache.Add(did, didDoc)
+
+	return didDoc, nil
+}
+
+var ErrAuditLogNotFound = errors.New("audit log not found for DID")
+
+func (c *PlcClient) GetDIDAuditLog(ctx context.Context, did string) (*DidAuditEntry, error) {
+	if val, ok := c.auditCache.Get(did); ok {
+		return val, nil
+	}
+
+	ustr := fmt.Sprintf("%s/%s/log/audit", c.plcHost, did)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, ustr, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create http request for DID audit log: %w", err)
+	}
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch DID audit log: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		io.Copy(io.Discard, resp.Body)
+		if resp.StatusCode == http.StatusNotFound {
+			return nil, ErrAuditLogNotFound
+		}
+		return nil, fmt.Errorf("DID audit log fetch returned unexpected status: %s", resp.Status)
+	}
+
+	var entries []DidAuditEntry
+	if err := json.NewDecoder(resp.Body).Decode(&entries); err != nil {
+		return nil, fmt.Errorf("failed to read DID audit log response bytes into DidAuditLog: %w", err)
+	}
+
+	if len(entries) == 0 {
+		return nil, fmt.Errorf("empty did audit log for did %s", did)
+	}
+
+	entry := entries[0]
+
+	if c.auditCache != nil {
+		c.auditCache.Add(did, &entry)
+	}
+
+	return &entry, nil
+}

cmd/atkafka/main.go

···
 				EnvVars: []string{"ATKAFKA_OSPREY_COMPATIBLE"},
 				Value:   false,
 			},
+			&cli.StringFlag{
+				Name:    "plc-host",
+				EnvVars: []string{"ATKAFKA_PLC_HOST"},
+			},
 		},
 		Action: func(cmd *cli.Context) error {
 			ctx := context.Background()
···
 				BootstrapServers: cmd.StringSlice("bootstrap-servers"),
 				OutputTopic:      cmd.String("output-topic"),
 				OspreyCompat:     cmd.Bool("osprey-compatible"),
+				PlcHost:          cmd.String("plc-host"),
 				Logger:           logger,
 			})
 

docker-compose.yml

···
       ATKAFKA_BOOTSTRAP_SERVERS: "kafka:29092"
       ATKAFKA_OUTPUT_TOPIC: "atproto-events"
       ATKAFKA_OSPREY_COMPATIBLE: "false"
+      ATKAFKA_PLC_HOST: "https://plc.directory"
     restart: unless-stopped
 
 volumes:

go.mod

···
 	github.com/bluesky-social/go-util v0.0.0-20251012040650-2ebbf57f5934
 	github.com/bluesky-social/indigo v0.0.0-20251125184450-35c1e15d2e5f
 	github.com/gorilla/websocket v1.5.3
+	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/joho/godotenv v1.5.1
 	github.com/prometheus/client_golang v1.23.2
 	github.com/twmb/franz-go v1.19.5
 	github.com/urfave/cli/v2 v2.25.7
+	golang.org/x/time v0.6.0
 )
 
 require (
···
 	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
 	github.com/hashicorp/go-retryablehttp v0.7.7 // indirect
 	github.com/hashicorp/golang-lru v1.0.2 // indirect
-	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/ipfs/bbloom v0.0.4 // indirect
 	github.com/ipfs/go-block-format v0.2.0 // indirect
 	github.com/ipfs/go-blockservice v0.5.2 // indirect
···
 	golang.org/x/sync v0.16.0 // indirect
 	golang.org/x/sys v0.35.0 // indirect
 	golang.org/x/text v0.28.0 // indirect
-	golang.org/x/time v0.6.0 // indirect
 	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
 	google.golang.org/protobuf v1.36.9 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect