diff --git a/cmd/bluepages/Dockerfile b/cmd/bluepages/Dockerfile new file mode 100644 --- /dev/null +++ b/cmd/bluepages/Dockerfile @@ -0,0 +1,37 @@ +# Run this dockerfile from the top level of the indigo git repository like: +# +# podman build -f ./cmd/bluepages/Dockerfile -t bluepages . + +### Compile stage +FROM golang:1.23-alpine3.20 AS build-env +RUN apk add --no-cache build-base make git + +ADD . /dockerbuild +WORKDIR /dockerbuild + +# timezone data for alpine builds +ENV GOEXPERIMENT=loopvar +RUN GIT_VERSION=$(git describe --tags --long --always) && \ + go build -tags timetzdata -o /bluepages ./cmd/bluepages + +### Run stage +FROM alpine:3.20 + +RUN apk add --no-cache --update dumb-init ca-certificates +ENTRYPOINT ["dumb-init", "--"] + +WORKDIR / +RUN mkdir -p data/bluepages +COPY --from=build-env /bluepages / + +# small things to make golang binaries work well under alpine +ENV GODEBUG=netdns=go +ENV TZ=Etc/UTC + +EXPOSE 2210 + +CMD ["/bluepages", "run"] + +LABEL org.opencontainers.image.source=https://github.com/bluesky-social/indigo +LABEL org.opencontainers.image.description="atproto identity directory (bluepages)" +LABEL org.opencontainers.image.licenses=MIT diff --git a/cmd/bluepages/README.md b/cmd/bluepages/README.md new file mode 100644 --- /dev/null +++ b/cmd/bluepages/README.md @@ -0,0 +1,17 @@ + +bluepages: an atproto identity directory +======================================== + +This is a simple API server which caches atproto handle and DID resolution responses. It is useful when you have a bunch of services that do identity resolution, and you don't want duplicated caches. + +Available commands, flags, and config are documented in the usage (`--help`). + +Current features and design decisions: + +- all caches stored in Redis +- will consume from the firehose (but doesn't yet) +- Lexicon API endpoints: + - `GET com.atproto.identity.resolveHandle` + - `GET com.atproto.identity.resolveDid` + - `GET com.atproto.identity.resolveIdentity` + - `POST com.atproto.identity.refreshIdentity` (admin auth) diff --git a/cmd/bluepages/firehose.go b/cmd/bluepages/firehose.go new file mode 100644 --- /dev/null +++ b/cmd/bluepages/firehose.go @@ -0,0 +1,152 @@ +package main + +import ( + "context" + "fmt" + "net/http" + "net/url" + "sync/atomic" + "time" + + comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/bluesky-social/indigo/events/schedulers/parallel" + + "github.com/bluesky-social/indigo/events" + "github.com/carlmjohnson/versioninfo" + "github.com/gorilla/websocket" + "github.com/redis/go-redis/v9" +) + +var firehoseCursorKey = "bluepages/firehoseSeq" + +func (srv *Server) RunFirehoseConsumer(ctx context.Context, host string, parallelism int) error { + + cur, err := srv.ReadLastCursor(ctx) + if err != nil { + return err + } + + dialer := websocket.DefaultDialer + u, err := url.Parse(host) + if err != nil { + return fmt.Errorf("invalid Host URI: %w", err) + } + u.Path = "xrpc/com.atproto.sync.subscribeRepos" + if cur != 0 { + u.RawQuery = fmt.Sprintf("cursor=%d", cur) + } + srv.logger.Info("subscribing to repo event stream", "upstream", host, "cursor", cur) + con, _, err := dialer.Dial(u.String(), http.Header{ + "User-Agent": []string{fmt.Sprintf("bluepages/%s", versioninfo.Short())}, + }) + if err != nil { + return fmt.Errorf("subscribing to firehose failed (dialing): %w", err) + } + + rsc := &events.RepoStreamCallbacks{ + RepoIdentity: func(evt *comatproto.SyncSubscribeRepos_Identity) error { + atomic.StoreInt64(&srv.lastSeq, evt.Seq) + ctx := context.Background() + srv.logger.Info("flushing cache due to #identity firehose event", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err) + + did, err := syntax.ParseDID(evt.Did) + if err != nil { + srv.logger.Warn("invalid DID in #identity event", "did", evt.Did, "seq", evt.Seq, "err", err) + return nil + } + if err := srv.dir.PurgeDID(ctx, did); err != nil { + srv.logger.Error("failed to purge DID from cache", "did", evt.Did, "seq", evt.Seq, "err", err) + return nil + } + if evt.Handle == nil { + return nil + } + handle, err := syntax.ParseHandle(*evt.Handle) + if err != nil { + srv.logger.Warn("invalid handle in #identity event", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err) + return nil + } + if err := srv.dir.PurgeHandle(ctx, handle); err != nil { + srv.logger.Error("failed to purge handle from cache", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err) + return nil + } + return nil + }, + } + + var scheduler events.Scheduler + // use a fixed-parallelism scheduler if configured + scheduler = parallel.NewScheduler( + parallelism, + 1000, + host, + rsc.EventHandler, + ) + srv.logger.Info("bluepages firehose scheduler configured", "scheduler", "parallel", "initial", parallelism) + + return events.HandleRepoStream(ctx, con, scheduler, srv.logger) +} + +func (srv *Server) ReadLastCursor(ctx context.Context) (int64, error) { + // if redis isn't configured, just skip + if srv.redisClient == nil { + srv.logger.Info("redis not configured, skipping cursor read") + return 0, nil + } + + val, err := srv.redisClient.Get(ctx, firehoseCursorKey).Int64() + if err == redis.Nil { + srv.logger.Info("no pre-existing cursor in redis") + return 0, nil + } else if err != nil { + return 0, err + } + srv.logger.Info("successfully found prior subscription cursor seq in redis", "seq", val) + return val, nil +} + +func (srv *Server) PersistCursor(ctx context.Context) error { + // if redis isn't configured, just skip + if srv.redisClient == nil { + return nil + } + lastSeq := atomic.LoadInt64(&srv.lastSeq) + if lastSeq <= 0 { + return nil + } + err := srv.redisClient.Set(ctx, firehoseCursorKey, lastSeq, 14*24*time.Hour).Err() + return err +} + +// this method runs in a loop, persisting the current cursor state every 5 seconds +func (srv *Server) RunPersistCursor(ctx context.Context) error { + + // if redis isn't configured, just skip + if srv.redisClient == nil { + return nil + } + ticker := time.NewTicker(5 * time.Second) + for { + select { + case <-ctx.Done(): + lastSeq := atomic.LoadInt64(&srv.lastSeq) + if lastSeq >= 1 { + srv.logger.Info("persisting final cursor seq value", "seq", lastSeq) + err := srv.PersistCursor(ctx) + if err != nil { + srv.logger.Error("failed to persist cursor", "err", err, "seq", lastSeq) + } + } + return nil + case <-ticker.C: + lastSeq := atomic.LoadInt64(&srv.lastSeq) + if lastSeq >= 1 { + err := srv.PersistCursor(ctx) + if err != nil { + srv.logger.Error("failed to persist cursor", "err", err, "seq", lastSeq) + } + } + } + } +} diff --git a/cmd/bluepages/handlers.go b/cmd/bluepages/handlers.go new file mode 100644 --- /dev/null +++ b/cmd/bluepages/handlers.go @@ -0,0 +1,258 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + + comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/identity" + "github.com/bluesky-social/indigo/atproto/syntax" + + "github.com/labstack/echo/v4" +) + +// GET /xrpc/com.atproto.identity.resolveHandle +func (srv *Server) ResolveHandle(c echo.Context) error { + ctx := c.Request().Context() + + hdl, err := syntax.ParseHandle(c.QueryParam("handle")) + if err != nil { + return c.JSON(400, GenericError{ + Error: "InvalidHandleSyntax", + Message: err.Error(), + }) + } + + did, err := srv.dir.ResolveHandle(ctx, hdl) + if err != nil && errors.Is(err, identity.ErrHandleNotFound) { + return c.JSON(404, GenericError{ + Error: "HandleNotFound", + Message: err.Error(), + }) + } else if err != nil { + return c.JSON(500, GenericError{ + Error: "InternalError", + Message: err.Error(), + }) + } + return c.JSON(200, comatproto.IdentityResolveHandle_Output{ + Did: did.String(), + }) +} + +// GET /xrpc/com.atproto.identity.resolveDid +func (srv *Server) ResolveDid(c echo.Context) error { + ctx := c.Request().Context() + + did, err := syntax.ParseDID(c.QueryParam("did")) + if err != nil { + return c.JSON(400, GenericError{ + Error: "InvalidDidSyntax", + Message: err.Error(), + }) + } + + rawDoc, err := srv.dir.ResolveDIDRaw(ctx, did) + if err != nil && errors.Is(err, identity.ErrDIDNotFound) { + return c.JSON(404, GenericError{ + Error: "DidNotFound", + Message: err.Error(), + }) + } else if err != nil { + return c.JSON(500, GenericError{ + Error: "InternalError", + Message: err.Error(), + }) + } + return c.JSON(200, comatproto.IdentityResolveDid_Output{ + DidDoc: rawDoc, + }) +} + +// helper for resolveIdentity +func (srv *Server) resolveIdentityFromHandle(c echo.Context, handle syntax.Handle) error { + ctx := c.Request().Context() + + did, err := srv.dir.ResolveHandle(ctx, handle) + if err != nil && errors.Is(err, identity.ErrHandleNotFound) { + return c.JSON(404, GenericError{ + Error: "HandleNotFound", + Message: err.Error(), + }) + } else if err != nil { + srv.logger.Warn("failed handle resolution", "err", err, "handle", handle) + return c.JSON(502, GenericError{ + Error: "HandleResolutionFailed", + Message: err.Error(), + }) + } + + rawDoc, err := srv.dir.ResolveDIDRaw(ctx, did) + if err != nil && errors.Is(err, identity.ErrDIDNotFound) { + return c.JSON(404, GenericError{ + Error: "DidNotFound", + Message: err.Error(), + }) + } else if err != nil { + return c.JSON(502, GenericError{ + Error: "DIDResolutionFailed", + Message: err.Error(), + }) + } + + var doc identity.DIDDocument + if err := json.Unmarshal(rawDoc, &doc); err != nil { + return c.JSON(400, GenericError{ + Error: "InvalidDidDocument", + Message: err.Error(), + }) + } + + ident := identity.ParseIdentity(&doc) + declHandle, err := ident.DeclaredHandle() + if err != nil || declHandle != handle { + return c.JSON(400, GenericError{ + Error: "HandleMismatch", + Message: err.Error(), + }) + } + + return c.JSON(200, comatproto.IdentityDefs_IdentityInfo{ + Did: ident.DID.String(), + Handle: handle.String(), + DidDoc: rawDoc, + }) +} + +// helper for resolveIdentity +func (srv *Server) resolveIdentityFromDID(c echo.Context, did syntax.DID) error { + ctx := c.Request().Context() + + rawDoc, err := srv.dir.ResolveDIDRaw(ctx, did) + if err != nil && errors.Is(err, identity.ErrDIDNotFound) { + return c.JSON(404, GenericError{ + Error: "DidNotFound", + Message: err.Error(), + }) + } else if err != nil { + return c.JSON(502, GenericError{ + Error: "DIDResolutionFailed", + Message: err.Error(), + }) + } + + var doc identity.DIDDocument + if err := json.Unmarshal(rawDoc, &doc); err != nil { + return c.JSON(400, GenericError{ + Error: "InvalidDidDocument", + Message: err.Error(), + }) + } + + ident := identity.ParseIdentity(&doc) + handle, err := ident.DeclaredHandle() + if err != nil { + // no handle declared, or invalid syntax + handle = syntax.Handle("handle.invalid") + } + + checkDID, err := srv.dir.ResolveHandle(ctx, handle) + if err != nil || checkDID != did { + handle = syntax.Handle("handle.invalid") + } + + return c.JSON(200, comatproto.IdentityDefs_IdentityInfo{ + Did: ident.DID.String(), + Handle: handle.String(), + DidDoc: rawDoc, + }) +} + +// GET /xrpc/com.atproto.identity.resolveIdentity +func (srv *Server) ResolveIdentity(c echo.Context) error { + // we partially re-implement the "Lookup()" logic here, but returning the full DID document, not `identity.Identity` + atid, err := syntax.ParseAtIdentifier(c.QueryParam("identifier")) + if err != nil { + return c.JSON(400, GenericError{ + Error: "InvalidIdentifierSyntax", + Message: err.Error(), + }) + } + + handle, err := atid.AsHandle() + if nil == err { + return srv.resolveIdentityFromHandle(c, handle) + } + did, err := atid.AsDID() + if nil == err { + return srv.resolveIdentityFromDID(c, did) + } + return fmt.Errorf("unreachable code path") +} + +// POST /xrpc/com.atproto.identity.refreshIdentity +func (srv *Server) RefreshIdentity(c echo.Context) error { + ctx := c.Request().Context() + + var body comatproto.IdentityRefreshIdentity_Input + if err := c.Bind(&body); err != nil { + return c.JSON(400, GenericError{ + Error: "InvalidRequestBody", + Message: err.Error(), + }) + } + + atid, err := syntax.ParseAtIdentifier(body.Identifier) + if err != nil { + return c.JSON(400, GenericError{ + Error: "InvalidIdentifierSyntax", + Message: err.Error(), + }) + } + + did, err := atid.AsDID() + if nil == err { + if err := srv.dir.PurgeDID(ctx, did); err != nil { + return err + } + return srv.resolveIdentityFromDID(c, did) + } + handle, err := atid.AsHandle() + if nil == err { + if err := srv.dir.PurgeHandle(ctx, handle); err != nil { + return err + } + return srv.resolveIdentityFromHandle(c, handle) + } + + return fmt.Errorf("unreachable code path") +} + +type GenericStatus struct { + Daemon string `json:"daemon"` + Status string `json:"status"` + Message string `json:"msg,omitempty"` +} + +func (s *Server) HandleHealthCheck(c echo.Context) error { + return c.JSON(200, GenericStatus{Status: "ok", Daemon: "bluepages"}) +} + +func (srv *Server) WebHome(c echo.Context) error { + return c.String(200, ` +eeeee e e e eeee eeeee eeeee eeeee eeee eeeee +8 8 8 8 8 8 8 8 8 8 8 8 8 8 " +8eee8e 8e 8e 8 8eee 8eee8 8eee8 8e 8eee 8eeee +88 8 88 88 8 88 88 88 8 88 "8 88 88 +88eee8 88eee 88ee8 88ee 88 88 8 88ee8 88ee 8ee88 + +This is an AT Protocol Identity Service + +Most API routes are under /xrpc/ + + Code: https://github.com/bluesky-social/indigo/tree/main/cmd/bluepages + Protocol: https://atproto.com + `) + +} diff --git a/cmd/bluepages/main.go b/cmd/bluepages/main.go new file mode 100644 --- /dev/null +++ b/cmd/bluepages/main.go @@ -0,0 +1,340 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + _ "net/http/pprof" + "os" + "runtime" + "strings" + + "github.com/bluesky-social/indigo/atproto/identity/apidir" + "github.com/bluesky-social/indigo/atproto/syntax" + + "github.com/carlmjohnson/versioninfo" + _ "github.com/joho/godotenv/autoload" + "github.com/urfave/cli/v2" +) + +func main() { + if err := run(os.Args); err != nil { + slog.Error("exiting", "err", err) + os.Exit(-1) + } +} + +func run(args []string) error { + + app := cli.App{ + Name: "bluepages", + Usage: "atproto identity directory", + Version: versioninfo.Short(), + Flags: []cli.Flag{ + &cli.StringFlag{ + Name: "atp-relay-host", + Usage: "hostname and port of Relay to subscribe to", + Value: "wss://bsky.network", + EnvVars: []string{"ATP_RELAY_HOST", "ATP_BGS_HOST"}, + }, + &cli.StringFlag{ + Name: "atp-plc-host", + Usage: "method, hostname, and port of PLC registry", + Value: "https://plc.directory", + EnvVars: []string{"ATP_PLC_HOST"}, + }, + &cli.IntFlag{ + Name: "plc-rate-limit", + Usage: "max number of requests per second to PLC registry", + Value: 300, + EnvVars: []string{"BLUEPAGES_PLC_RATE_LIMIT"}, + }, + &cli.StringFlag{ + Name: "redis-url", + Usage: "redis connection URL: redis://:@:6379/", + Value: "redis://localhost:6379/0", + EnvVars: []string{"BLUEPAGES_REDIS_URL"}, + }, + &cli.StringFlag{ + Name: "log-level", + Usage: "log verbosity level (eg: warn, info, debug)", + EnvVars: []string{"BLUEPAGES_LOG_LEVEL", "GO_LOG_LEVEL", "LOG_LEVEL"}, + }, + }, + Commands: []*cli.Command{ + &cli.Command{ + Name: "serve", + Usage: "run the bluepages API daemon", + Action: runServeCmd, + Flags: []cli.Flag{ + &cli.StringFlag{ + Name: "bind", + Usage: "Specify the local IP/port to bind to", + Required: false, + Value: ":6600", + EnvVars: []string{"BLUEPAGES_BIND"}, + }, + &cli.StringFlag{ + Name: "metrics-listen", + Usage: "IP or address, and port, to listen on for metrics APIs", + Value: ":3989", + EnvVars: []string{"BLUEPAGES_METRICS_LISTEN"}, + }, + &cli.BoolFlag{ + Name: "disable-firehose-consumer", + Usage: "don't consume #identity events from firehose", + EnvVars: []string{"BLUEPAGES_DISABLE_FIREHOSE_CONSUMER"}, + }, + &cli.BoolFlag{ + Name: "disable-refresh", + Usage: "disable the refreshIdentity API endpoint", + EnvVars: []string{"BLUEPAGES_DISABLE_REFRESH"}, + }, + &cli.IntFlag{ + Name: "firehose-parallelism", + Usage: "number of concurrent firehose workers", + Value: 4, + EnvVars: []string{"BLUEPAGES_FIREHOSE_PARALLELISM"}, + }, + }, + }, + &cli.Command{ + Name: "resolve-handle", + ArgsUsage: ``, + Usage: "query service for handle resoltion", + Action: runResolveHandleCmd, + Flags: []cli.Flag{ + &cli.StringFlag{ + Name: "host", + Usage: "bluepages server to send request to", + Value: "http://localhost:6600", + EnvVars: []string{"BLUEPAGES_HOST"}, + }, + }, + }, + &cli.Command{ + Name: "resolve-did", + ArgsUsage: ``, + Usage: "query service for DID document resoltion", + Action: runResolveDIDCmd, + Flags: []cli.Flag{ + &cli.StringFlag{ + Name: "host", + Usage: "bluepages server to send request to", + Value: "http://localhost:6600", + EnvVars: []string{"BLUEPAGES_HOST"}, + }, + }, + }, + &cli.Command{ + Name: "lookup", + ArgsUsage: ``, + Usage: "query service for identity resoltion", + Action: runLookupCmd, + Flags: []cli.Flag{ + &cli.StringFlag{ + Name: "host", + Usage: "bluepages server to send request to", + Value: "http://localhost:6600", + EnvVars: []string{"BLUEPAGES_HOST"}, + }, + }, + }, + &cli.Command{ + Name: "refresh", + ArgsUsage: ``, + Usage: "ask service to refresh identity", + Action: runRefreshCmd, + Flags: []cli.Flag{ + &cli.StringFlag{ + Name: "host", + Usage: "bluepages server to send request to", + Value: "http://localhost:6600", + EnvVars: []string{"BLUEPAGES_HOST"}, + }, + }, + }, + }, + } + + return app.Run(args) +} + +func configLogger(cctx *cli.Context, writer io.Writer) *slog.Logger { + var level slog.Level + switch strings.ToLower(cctx.String("log-level")) { + case "error": + level = slog.LevelError + case "warn": + level = slog.LevelWarn + case "info": + level = slog.LevelInfo + case "debug": + level = slog.LevelDebug + default: + level = slog.LevelInfo + } + logger := slog.New(slog.NewJSONHandler(writer, &slog.HandlerOptions{ + Level: level, + })) + slog.SetDefault(logger) + return logger +} + +func configClient(cctx *cli.Context) apidir.APIDirectory { + return apidir.NewAPIDirectory(cctx.String("host")) +} + +func runServeCmd(cctx *cli.Context) error { + logger := configLogger(cctx, os.Stdout) + ctx := context.Background() + + srv, err := NewServer( + Config{ + Logger: logger, + Bind: cctx.String("bind"), + RedisURL: cctx.String("redis-url"), + PLCHost: cctx.String("atp-plc-host"), + PLCRateLimit: cctx.Int("plc-rate-limit"), + DisableRefresh: cctx.Bool("disable-refresh"), + }, + ) + if err != nil { + return fmt.Errorf("failed to construct server: %v", err) + } + + if !cctx.Bool("disable-firehose-consumer") { + go func() { + firehoseHost := cctx.String("atp-relay-host") + firehoseParallelism := cctx.Int("firehose-parallelism") + if err := srv.RunFirehoseConsumer(ctx, firehoseHost, firehoseParallelism); err != nil { + slog.Error("firehose consumer thread failed", "err", err) + // NOTE: not crashing or halting process here + } + }() + go func() { + if err := srv.RunPersistCursor(ctx); err != nil { + slog.Error("firehose persist thread failed", "err", err) + // NOTE: not crashing or halting process here + } + }() + } + + // prometheus HTTP endpoint: /metrics + go func() { + // TODO: what is this tuning for? just cargo-culted it + runtime.SetBlockProfileRate(10) + runtime.SetMutexProfileFraction(10) + if err := srv.RunMetrics(cctx.String("metrics-listen")); err != nil { + slog.Error("failed to start metrics endpoint", "error", err) + // NOTE: not crashing or halting process here + } + }() + + return srv.RunAPI() +} + +func runResolveHandleCmd(cctx *cli.Context) error { + ctx := context.Background() + dir := configClient(cctx) + + s := cctx.Args().First() + if s == "" { + return fmt.Errorf("need to provide identifier for resolution") + } + handle, err := syntax.ParseHandle(s) + if err != nil { + return err + } + + did, err := dir.ResolveHandle(ctx, handle) + if err != nil { + return err + } + fmt.Println(did.String()) + return nil +} + +func runResolveDIDCmd(cctx *cli.Context) error { + ctx := context.Background() + dir := configClient(cctx) + + s := cctx.Args().First() + if s == "" { + return fmt.Errorf("need to provide identifier for resolution") + } + did, err := syntax.ParseDID(s) + if err != nil { + return err + } + + raw, err := dir.ResolveDIDRaw(ctx, did) + if err != nil { + return err + } + b, err := json.MarshalIndent(raw, "", " ") + if err != nil { + return err + } + fmt.Println(string(b)) + return nil +} + +func runLookupCmd(cctx *cli.Context) error { + ctx := context.Background() + dir := configClient(cctx) + + s := cctx.Args().First() + if s == "" { + return fmt.Errorf("need to provide identifier for resolution") + } + atid, err := syntax.ParseAtIdentifier(s) + if err != nil { + return err + } + + ident, err := dir.Lookup(ctx, *atid) + if err != nil { + return err + } + + b, err := json.MarshalIndent(ident, "", " ") + if err != nil { + return err + } + fmt.Println(string(b)) + return nil +} + +func runRefreshCmd(cctx *cli.Context) error { + ctx := context.Background() + dir := configClient(cctx) + + s := cctx.Args().First() + if s == "" { + return fmt.Errorf("need to provide identifier for resolution") + } + atid, err := syntax.ParseAtIdentifier(s) + if err != nil { + return err + } + + err = dir.Purge(ctx, *atid) + if err != nil { + return err + } + + ident, err := dir.Lookup(ctx, *atid) + if err != nil { + return err + } + + b, err := json.MarshalIndent(ident, "", " ") + if err != nil { + return err + } + fmt.Println(string(b)) + return nil +} diff --git a/cmd/bluepages/metrics.go b/cmd/bluepages/metrics.go new file mode 100644 --- /dev/null +++ b/cmd/bluepages/metrics.go @@ -0,0 +1,58 @@ +package main + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +var handleCacheHits = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bluepages_resolve_handle_cache_hits", + Help: "Number of cache hits for ATProto handle resolutions", +}) + +var handleCacheMisses = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bluepages_resolve_handle_cache_misses", + Help: "Number of cache misses for ATProto handle resolutions", +}) + +var handleRequestsCoalesced = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bluepages_resolve_handle_requests_coalesced", + Help: "Number of handle requests coalesced", +}) + +var handleResolutionErrors = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bluepages_resolve_handle_resolution_errors", + Help: "Number of non-cached handle resolution errors", +}) + +var handleResolveDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ + Name: "bluepages_resolve_handle_duration", + Help: "Time to resolve a handle from network (not cached)", + Buckets: prometheus.ExponentialBucketsRange(0.001, 2, 15), +}, []string{"status"}) + +var didCacheHits = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bluepages_resolve_did_cache_hits", + Help: "Number of cache hits for ATProto DID resolutions", +}) + +var didCacheMisses = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bluepages_resolve_did_cache_misses", + Help: "Number of cache misses for ATProto DID resolutions", +}) + +var didRequestsCoalesced = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bluepages_resolve_did_requests_coalesced", + Help: "Number of DID requests coalesced", +}) + +var didResolutionErrors = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bluepages_resolve_did_resolution_errors", + Help: "Number of non-cached DID resolution errors", +}) + +var didResolveDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ + Name: "bluepages_resolve_did_duration", + Help: "Time to resolve a DID from network (not cached)", + Buckets: prometheus.ExponentialBucketsRange(0.001, 2, 15), +}, []string{"status"}) diff --git a/cmd/bluepages/resolver.go b/cmd/bluepages/resolver.go new file mode 100644 --- /dev/null +++ b/cmd/bluepages/resolver.go @@ -0,0 +1,319 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "sync" + "time" + + "github.com/bluesky-social/indigo/atproto/identity" + "github.com/bluesky-social/indigo/atproto/syntax" + + "github.com/go-redis/cache/v9" + "github.com/redis/go-redis/v9" +) + +// This file is a fork of indigo:atproto/identity/redisdir. It stores raw DID documents, not identities, and implements `identity.Resolver`. + +// Uses redis as a cache for identity lookups. +// +// Includes an in-process LRU cache as well (provided by the redis client library), for hot key (identities). +type RedisResolver struct { + Inner identity.Resolver + ErrTTL time.Duration + HitTTL time.Duration + InvalidHandleTTL time.Duration + Logger *slog.Logger + + handleCache *cache.Cache + didCache *cache.Cache + didResolveChans sync.Map + handleResolveChans sync.Map +} + +type handleEntry struct { + Updated time.Time + // needs to be pointer type, because unmarshalling empty string would be an error + DID *syntax.DID + Err error +} + +type didEntry struct { + Updated time.Time + RawDoc json.RawMessage + Err error +} + +var _ identity.Resolver = (*RedisResolver)(nil) + +// Creates a new caching `identity.Resolver` wrapper around an existing directory, using Redis and in-process LRU for caching. +// +// `redisURL` contains all the redis connection config options. +// `hitTTL` and `errTTL` define how long successful and errored identity metadata should be cached (respectively). errTTL is expected to be shorted than hitTTL. +// `lruSize` is the size of the in-process cache, for each of the handle and identity caches. 10000 is a reasonable default. +// +// NOTE: Errors returned may be inconsistent with the base directory, or between calls. This is because cached errors are serialized/deserialized and that may break equality checks. +func NewRedisResolver(inner identity.Resolver, redisURL string, hitTTL, errTTL, invalidHandleTTL time.Duration, lruSize int) (*RedisResolver, error) { + opt, err := redis.ParseURL(redisURL) + if err != nil { + return nil, fmt.Errorf("could not configure redis identity cache: %w", err) + } + rdb := redis.NewClient(opt) + // check redis connection + _, err = rdb.Ping(context.TODO()).Result() + if err != nil { + return nil, fmt.Errorf("could not connect to redis identity cache: %w", err) + } + handleCache := cache.New(&cache.Options{ + Redis: rdb, + LocalCache: cache.NewTinyLFU(lruSize, hitTTL), + }) + didCache := cache.New(&cache.Options{ + Redis: rdb, + LocalCache: cache.NewTinyLFU(lruSize, hitTTL), + }) + return &RedisResolver{ + Inner: inner, + ErrTTL: errTTL, + HitTTL: hitTTL, + InvalidHandleTTL: invalidHandleTTL, + handleCache: handleCache, + didCache: didCache, + }, nil +} + +func (d *RedisResolver) isHandleStale(e *handleEntry) bool { + if e.Err != nil && time.Since(e.Updated) > d.ErrTTL { + return true + } + return false +} + +func (d *RedisResolver) isDIDStale(e *didEntry) bool { + if e.Err != nil && time.Since(e.Updated) > d.ErrTTL { + return true + } + return false +} + +func (d *RedisResolver) refreshHandle(ctx context.Context, h syntax.Handle) handleEntry { + start := time.Now() + did, err := d.Inner.ResolveHandle(ctx, h) + duration := time.Since(start) + + if err != nil { + d.Logger.Info("handle resolution failed", "handle", h, "duration", duration, "err", err) + handleResolutionErrors.Inc() + handleResolveDuration.WithLabelValues("fail").Observe(time.Since(start).Seconds()) + } else { + handleResolveDuration.WithLabelValues("success").Observe(time.Since(start).Seconds()) + } + if duration.Seconds() > 5.0 { + d.Logger.Info("slow handle resolution", "handle", h, "duration", duration) + } + + he := handleEntry{ + Updated: time.Now(), + DID: &did, + Err: err, + } + err = d.handleCache.Set(&cache.Item{ + Ctx: ctx, + Key: "bluepages/handle/" + h.String(), + Value: he, + TTL: d.ErrTTL, + }) + if err != nil { + d.Logger.Error("identity cache write failed", "cache", "handle", "err", err) + } + return he +} + +func (d *RedisResolver) refreshDID(ctx context.Context, did syntax.DID) didEntry { + start := time.Now() + rawDoc, err := d.Inner.ResolveDIDRaw(ctx, did) + duration := time.Since(start) + + if err != nil { + d.Logger.Info("DID resolution failed", "did", did, "duration", duration, "err", err) + didResolutionErrors.Inc() + didResolveDuration.WithLabelValues("fail").Observe(time.Since(start).Seconds()) + } else { + didResolveDuration.WithLabelValues("success").Observe(time.Since(start).Seconds()) + } + if duration.Seconds() > 5.0 { + d.Logger.Info("slow DID resolution", "did", did, "duration", duration) + } + + // persist the DID lookup error, instead of processing it immediately + entry := didEntry{ + Updated: time.Now(), + RawDoc: rawDoc, + Err: err, + } + + err = d.didCache.Set(&cache.Item{ + Ctx: ctx, + Key: "bluepages/did/" + did.String(), + Value: entry, + TTL: d.HitTTL, + }) + if err != nil { + d.Logger.Error("DID cache write failed", "cache", "did", "did", did, "err", err) + } + return entry +} + +func (d *RedisResolver) ResolveHandle(ctx context.Context, h syntax.Handle) (syntax.DID, error) { + if h.IsInvalidHandle() { + return "", fmt.Errorf("can not resolve handle: %w", identity.ErrInvalidHandle) + } + h = h.Normalize() + var entry handleEntry + err := d.handleCache.Get(ctx, "bluepages/handle/"+h.String(), &entry) + if err != nil && err != cache.ErrCacheMiss { + return "", fmt.Errorf("identity cache read failed: %w", err) + } + if err == nil && !d.isHandleStale(&entry) { // if no error... + handleCacheHits.Inc() + if entry.Err != nil { + return "", entry.Err + } else if entry.DID != nil { + return *entry.DID, nil + } else { + return "", errors.New("code flow error in redis identity directory") + } + } + handleCacheMisses.Inc() + + // Coalesce multiple requests for the same Handle + res := make(chan struct{}) + val, loaded := d.handleResolveChans.LoadOrStore(h.String(), res) + if loaded { + handleRequestsCoalesced.Inc() + // Wait for the result from the pending request + select { + case <-val.(chan struct{}): + // The result should now be in the cache + err := d.handleCache.Get(ctx, "bluepages/handle/"+h.String(), entry) + if err != nil && err != cache.ErrCacheMiss { + return "", fmt.Errorf("identity cache read failed: %w", err) + } + if err == nil && !d.isHandleStale(&entry) { // if no error... + if entry.Err != nil { + return "", entry.Err + } else if entry.DID != nil { + return *entry.DID, nil + } else { + return "", errors.New("code flow error in redis identity directory") + } + } + return "", errors.New("identity not found in cache after coalesce returned") + case <-ctx.Done(): + return "", ctx.Err() + } + } + + // Update the Handle Entry from PLC and cache the result + newEntry := d.refreshHandle(ctx, h) + + // Cleanup the coalesce map and close the results channel + d.handleResolveChans.Delete(h.String()) + // Callers waiting will now get the result from the cache + close(res) + + if newEntry.Err != nil { + return "", newEntry.Err + } + if newEntry.DID != nil { + return *newEntry.DID, nil + } + return "", errors.New("unexpected control-flow error") +} + +func (d *RedisResolver) ResolveDIDRaw(ctx context.Context, did syntax.DID) (json.RawMessage, error) { + var entry didEntry + err := d.didCache.Get(ctx, "bluepages/did/"+did.String(), &entry) + if err != nil && err != cache.ErrCacheMiss { + return nil, fmt.Errorf("DID cache read failed: %w", err) + } + if err == nil && !d.isDIDStale(&entry) { // if no error... + didCacheHits.Inc() + return entry.RawDoc, entry.Err + } + didCacheMisses.Inc() + + // Coalesce multiple requests for the same DID + res := make(chan struct{}) + val, loaded := d.didResolveChans.LoadOrStore(did.String(), res) + if loaded { + didRequestsCoalesced.Inc() + // Wait for the result from the pending request + select { + case <-val.(chan struct{}): + // The result should now be in the cache + err = d.didCache.Get(ctx, "bluepages/did/"+did.String(), &entry) + if err != nil && err != cache.ErrCacheMiss { + return nil, fmt.Errorf("DID cache read failed: %w", err) + } + if err == nil && !d.isDIDStale(&entry) { // if no error... + return entry.RawDoc, entry.Err + } + return nil, errors.New("DID not found in cache after coalesce returned") + case <-ctx.Done(): + return nil, ctx.Err() + } + } + + // Update the DID Entry and cache the result + newEntry := d.refreshDID(ctx, did) + + // Cleanup the coalesce map and close the results channel + d.didResolveChans.Delete(did.String()) + // Callers waiting will now get the result from the cache + close(res) + + if newEntry.Err != nil { + return nil, newEntry.Err + } + if newEntry.RawDoc != nil { + return newEntry.RawDoc, nil + } + return nil, errors.New("unexpected control-flow error") +} + +func (d *RedisResolver) ResolveDID(ctx context.Context, did syntax.DID) (*identity.DIDDocument, error) { + b, err := d.ResolveDIDRaw(ctx, did) + if err != nil { + return nil, err + } + + var doc identity.DIDDocument + if err := json.Unmarshal(b, &doc); err != nil { + return nil, fmt.Errorf("%w: JSON DID document parse: %w", identity.ErrDIDResolutionFailed, err) + } + if doc.DID != did { + return nil, fmt.Errorf("document ID did not match DID") + } + return &doc, nil +} + +func (d *RedisResolver) PurgeHandle(ctx context.Context, handle syntax.Handle) error { + handle = handle.Normalize() + err := d.handleCache.Delete(ctx, "bluepages/handle/"+handle.String()) + if err == cache.ErrCacheMiss { + return nil + } + return err +} + +func (d *RedisResolver) PurgeDID(ctx context.Context, did syntax.DID) error { + err := d.didCache.Delete(ctx, "bluepages/did/"+did.String()) + if err == cache.ErrCacheMiss { + return nil + } + return err +} diff --git a/cmd/bluepages/server.go b/cmd/bluepages/server.go new file mode 100644 --- /dev/null +++ b/cmd/bluepages/server.go @@ -0,0 +1,218 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log/slog" + "net" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/bluesky-social/indigo/atproto/identity" + + "github.com/labstack/echo/v4" + "github.com/labstack/echo/v4/middleware" + "github.com/prometheus/client_golang/prometheus/promhttp" + "github.com/redis/go-redis/v9" + slogecho "github.com/samber/slog-echo" + "golang.org/x/time/rate" +) + +type Server struct { + dir *RedisResolver + echo *echo.Echo + httpd *http.Server + logger *slog.Logger + + // this redis client is used to store firehose offset + redisClient *redis.Client + + // lastSeq is the most recent event sequence number we've received and begun to handle. + // This number is periodically persisted to redis, if redis is present. + // The value is best-effort (the stream handling itself is concurrent, so event numbers may not be monotonic), + // but nonetheless, you must use atomics when updating or reading this (to avoid data races). + lastSeq int64 +} + +type Config struct { + Logger *slog.Logger + PLCHost string + PLCRateLimit int + RedisURL string + Bind string + DisableRefresh bool +} + +func NewServer(config Config) (*Server, error) { + logger := config.Logger + if logger == nil { + logger = slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{ + Level: slog.LevelInfo, + })) + } + + baseDir := identity.BaseDirectory{ + PLCURL: config.PLCHost, + HTTPClient: http.Client{ + Timeout: time.Second * 10, + Transport: &http.Transport{ + // would want this around 100ms for services doing lots of handle resolution (to reduce number of idle connections). Impacts PLC connections as well, but not too bad. + IdleConnTimeout: time.Millisecond * 100, + MaxIdleConns: 1000, + }, + }, + Resolver: net.Resolver{ + Dial: func(ctx context.Context, network, address string) (net.Conn, error) { + d := net.Dialer{Timeout: time.Second * 3} + return d.DialContext(ctx, network, address) + }, + }, + PLCLimiter: rate.NewLimiter(rate.Limit(config.PLCRateLimit), 1), + TryAuthoritativeDNS: true, + SkipDNSDomainSuffixes: []string{".bsky.social", ".staging.bsky.dev"}, + // TODO: UserAgent: "bluepages", + } + + // TODO: config these timeouts + redisDir, err := NewRedisResolver(&baseDir, config.RedisURL, time.Hour*24, time.Minute*2, time.Minute*5, 50_000) + if err != nil { + return nil, err + } + redisDir.Logger = logger + + // configure redis client (for firehose consumer) + redisOpt, err := redis.ParseURL(config.RedisURL) + if err != nil { + return nil, fmt.Errorf("parsing redis URL: %v", err) + } + redisClient := redis.NewClient(redisOpt) + // check redis connection + _, err = redisClient.Ping(context.Background()).Result() + if err != nil { + return nil, fmt.Errorf("redis ping failed: %v", err) + } + + e := echo.New() + + // httpd + var ( + httpTimeout = 1 * time.Minute + httpMaxHeaderBytes = 1 * (1024 * 1024) + ) + + srv := &Server{ + echo: e, + dir: redisDir, + logger: logger, + redisClient: redisClient, + } + + srv.httpd = &http.Server{ + Handler: srv, + Addr: config.Bind, + WriteTimeout: httpTimeout, + ReadTimeout: httpTimeout, + MaxHeaderBytes: httpMaxHeaderBytes, + } + + e.HideBanner = true + e.Use(slogecho.New(logger)) + e.Use(middleware.Recover()) + e.Use(middleware.BodyLimit("4M")) + e.HTTPErrorHandler = srv.errorHandler + e.Use(middleware.SecureWithConfig(middleware.SecureConfig{ + ContentTypeNosniff: "nosniff", + XFrameOptions: "SAMEORIGIN", + HSTSMaxAge: 31536000, // 365 days + // TODO: + // ContentSecurityPolicy + // XSSProtection + })) + + e.GET("/", srv.WebHome) + e.GET("/_health", srv.HandleHealthCheck) + e.GET("/xrpc/com.atproto.identity.resolveHandle", srv.ResolveHandle) + e.GET("/xrpc/com.atproto.identity.resolveDid", srv.ResolveDid) + e.GET("/xrpc/com.atproto.identity.resolveIdentity", srv.ResolveIdentity) + if !config.DisableRefresh { + e.POST("/xrpc/com.atproto.identity.refreshIdentity", srv.RefreshIdentity) + } + + return srv, nil +} + +func (srv *Server) ServeHTTP(rw http.ResponseWriter, req *http.Request) { + srv.echo.ServeHTTP(rw, req) +} + +func (srv *Server) RunAPI() error { + srv.logger.Info("starting server", "bind", srv.httpd.Addr) + go func() { + if err := srv.httpd.ListenAndServe(); err != nil { + if !errors.Is(err, http.ErrServerClosed) { + srv.logger.Error("HTTP server shutting down unexpectedly", "err", err) + } + } + }() + + // Wait for a signal to exit. + srv.logger.Info("registering OS exit signal handler") + quit := make(chan struct{}) + exitSignals := make(chan os.Signal, 1) + signal.Notify(exitSignals, syscall.SIGINT, syscall.SIGTERM) + go func() { + sig := <-exitSignals + srv.logger.Info("received OS exit signal", "signal", sig) + + // Shut down the HTTP server + if err := srv.Shutdown(); err != nil { + srv.logger.Error("HTTP server shutdown error", "err", err) + } + + // Trigger the return that causes an exit. + close(quit) + }() + <-quit + srv.logger.Info("graceful shutdown complete") + return nil +} + +func (srv *Server) RunMetrics(bind string) error { + p := "/metrics" + srv.logger.Info("starting metrics endpoint", "bind", bind, "path", p) + http.Handle(p, promhttp.Handler()) + return http.ListenAndServe(bind, nil) +} + +func (srv *Server) Shutdown() error { + srv.logger.Info("shutting down") + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + return srv.httpd.Shutdown(ctx) +} + +type GenericError struct { + Error string `json:"error"` + Message string `json:"message"` +} + +func (srv *Server) errorHandler(err error, c echo.Context) { + code := http.StatusInternalServerError + var errorMessage string + if he, ok := err.(*echo.HTTPError); ok { + code = he.Code + errorMessage = fmt.Sprintf("%s", he.Message) + } + if code >= 500 { + srv.logger.Warn("bluepages-http-internal-error", "err", err) + } + if !c.Response().Committed { + c.JSON(code, GenericError{Error: "InternalError", Message: errorMessage}) + } +} diff --git a/cmd/domesday/Dockerfile b/cmd/domesday/Dockerfile deleted file mode 100644 --- a/cmd/domesday/Dockerfile +++ /dev/null @@ -1,37 +0,0 @@ -# Run this dockerfile from the top level of the indigo git repository like: -# -# podman build -f ./cmd/domesday/Dockerfile -t domesday . - -### Compile stage -FROM golang:1.23-alpine3.20 AS build-env -RUN apk add --no-cache build-base make git - -ADD . /dockerbuild -WORKDIR /dockerbuild - -# timezone data for alpine builds -ENV GOEXPERIMENT=loopvar -RUN GIT_VERSION=$(git describe --tags --long --always) && \ - go build -tags timetzdata -o /domesday ./cmd/domesday - -### Run stage -FROM alpine:3.20 - -RUN apk add --no-cache --update dumb-init ca-certificates -ENTRYPOINT ["dumb-init", "--"] - -WORKDIR / -RUN mkdir -p data/domesday -COPY --from=build-env /domesday / - -# small things to make golang binaries work well under alpine -ENV GODEBUG=netdns=go -ENV TZ=Etc/UTC - -EXPOSE 2210 - -CMD ["/domesday", "run"] - -LABEL org.opencontainers.image.source=https://github.com/bluesky-social/indigo -LABEL org.opencontainers.image.description="atproto identity directory (domesday)" -LABEL org.opencontainers.image.licenses=MIT diff --git a/cmd/domesday/README.md b/cmd/domesday/README.md deleted file mode 100644 --- a/cmd/domesday/README.md +++ /dev/null @@ -1,19 +0,0 @@ - -domesday: an atproto identity directory -======================================== - -This is a simple API server which caches atproto handle and DID resolution responses. It is useful when you have a bunch of services that do identity resolution, and you don't want duplicated caches. - -The name is a reference to the [Domesday Book](https://en.wikipedia.org/wiki/Domesday_Book), an early manuscript recoding a English census in 1086. It is a big fancy book with a lot of names in it. - -Available commands, flags, and config are documented in the usage (`--help`). - -Current features and design decisions: - -- all caches stored in Redis -- will consume from the firehose (but doesn't yet) -- Lexicon API endpoints: - - `GET com.atproto.identity.resolveHandle` - - `GET com.atproto.identity.resolveDid` - - `GET com.atproto.identity.resolveIdentity` - - `POST com.atproto.identity.refreshIdentity` (admin auth) diff --git a/cmd/domesday/firehose.go b/cmd/domesday/firehose.go deleted file mode 100644 --- a/cmd/domesday/firehose.go +++ /dev/null @@ -1,152 +0,0 @@ -package main - -import ( - "context" - "fmt" - "net/http" - "net/url" - "sync/atomic" - "time" - - comatproto "github.com/bluesky-social/indigo/api/atproto" - "github.com/bluesky-social/indigo/atproto/syntax" - "github.com/bluesky-social/indigo/events/schedulers/parallel" - - "github.com/bluesky-social/indigo/events" - "github.com/carlmjohnson/versioninfo" - "github.com/gorilla/websocket" - "github.com/redis/go-redis/v9" -) - -var firehoseCursorKey = "domes/firehoseSeq" - -func (srv *Server) RunFirehoseConsumer(ctx context.Context, host string, parallelism int) error { - - cur, err := srv.ReadLastCursor(ctx) - if err != nil { - return err - } - - dialer := websocket.DefaultDialer - u, err := url.Parse(host) - if err != nil { - return fmt.Errorf("invalid Host URI: %w", err) - } - u.Path = "xrpc/com.atproto.sync.subscribeRepos" - if cur != 0 { - u.RawQuery = fmt.Sprintf("cursor=%d", cur) - } - srv.logger.Info("subscribing to repo event stream", "upstream", host, "cursor", cur) - con, _, err := dialer.Dial(u.String(), http.Header{ - "User-Agent": []string{fmt.Sprintf("domesday/%s", versioninfo.Short())}, - }) - if err != nil { - return fmt.Errorf("subscribing to firehose failed (dialing): %w", err) - } - - rsc := &events.RepoStreamCallbacks{ - RepoIdentity: func(evt *comatproto.SyncSubscribeRepos_Identity) error { - atomic.StoreInt64(&srv.lastSeq, evt.Seq) - ctx := context.Background() - srv.logger.Info("flushing cache due to #identity firehose event", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err) - - did, err := syntax.ParseDID(evt.Did) - if err != nil { - srv.logger.Warn("invalid DID in #identity event", "did", evt.Did, "seq", evt.Seq, "err", err) - return nil - } - if err := srv.dir.PurgeDID(ctx, did); err != nil { - srv.logger.Error("failed to purge DID from cache", "did", evt.Did, "seq", evt.Seq, "err", err) - return nil - } - if evt.Handle == nil { - return nil - } - handle, err := syntax.ParseHandle(*evt.Handle) - if err != nil { - srv.logger.Warn("invalid handle in #identity event", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err) - return nil - } - if err := srv.dir.PurgeHandle(ctx, handle); err != nil { - srv.logger.Error("failed to purge handle from cache", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err) - return nil - } - return nil - }, - } - - var scheduler events.Scheduler - // use a fixed-parallelism scheduler if configured - scheduler = parallel.NewScheduler( - parallelism, - 1000, - host, - rsc.EventHandler, - ) - srv.logger.Info("domesday firehose scheduler configured", "scheduler", "parallel", "initial", parallelism) - - return events.HandleRepoStream(ctx, con, scheduler, srv.logger) -} - -func (srv *Server) ReadLastCursor(ctx context.Context) (int64, error) { - // if redis isn't configured, just skip - if srv.redisClient == nil { - srv.logger.Info("redis not configured, skipping cursor read") - return 0, nil - } - - val, err := srv.redisClient.Get(ctx, firehoseCursorKey).Int64() - if err == redis.Nil { - srv.logger.Info("no pre-existing cursor in redis") - return 0, nil - } else if err != nil { - return 0, err - } - srv.logger.Info("successfully found prior subscription cursor seq in redis", "seq", val) - return val, nil -} - -func (srv *Server) PersistCursor(ctx context.Context) error { - // if redis isn't configured, just skip - if srv.redisClient == nil { - return nil - } - lastSeq := atomic.LoadInt64(&srv.lastSeq) - if lastSeq <= 0 { - return nil - } - err := srv.redisClient.Set(ctx, firehoseCursorKey, lastSeq, 14*24*time.Hour).Err() - return err -} - -// this method runs in a loop, persisting the current cursor state every 5 seconds -func (srv *Server) RunPersistCursor(ctx context.Context) error { - - // if redis isn't configured, just skip - if srv.redisClient == nil { - return nil - } - ticker := time.NewTicker(5 * time.Second) - for { - select { - case <-ctx.Done(): - lastSeq := atomic.LoadInt64(&srv.lastSeq) - if lastSeq >= 1 { - srv.logger.Info("persisting final cursor seq value", "seq", lastSeq) - err := srv.PersistCursor(ctx) - if err != nil { - srv.logger.Error("failed to persist cursor", "err", err, "seq", lastSeq) - } - } - return nil - case <-ticker.C: - lastSeq := atomic.LoadInt64(&srv.lastSeq) - if lastSeq >= 1 { - err := srv.PersistCursor(ctx) - if err != nil { - srv.logger.Error("failed to persist cursor", "err", err, "seq", lastSeq) - } - } - } - } -} diff --git a/cmd/domesday/handlers.go b/cmd/domesday/handlers.go deleted file mode 100644 --- a/cmd/domesday/handlers.go +++ /dev/null @@ -1,260 +0,0 @@ -package main - -import ( - "encoding/json" - "errors" - "fmt" - - comatproto "github.com/bluesky-social/indigo/api/atproto" - "github.com/bluesky-social/indigo/atproto/identity" - "github.com/bluesky-social/indigo/atproto/syntax" - - "github.com/labstack/echo/v4" -) - -// GET /xrpc/com.atproto.identity.resolveHandle -func (srv *Server) ResolveHandle(c echo.Context) error { - ctx := c.Request().Context() - - hdl, err := syntax.ParseHandle(c.QueryParam("handle")) - if err != nil { - return c.JSON(400, GenericError{ - Error: "InvalidHandleSyntax", - Message: err.Error(), - }) - } - - did, err := srv.dir.ResolveHandle(ctx, hdl) - if err != nil && errors.Is(err, identity.ErrHandleNotFound) { - return c.JSON(404, GenericError{ - Error: "HandleNotFound", - Message: err.Error(), - }) - } else if err != nil { - return c.JSON(500, GenericError{ - Error: "InternalError", - Message: err.Error(), - }) - } - return c.JSON(200, comatproto.IdentityResolveHandle_Output{ - Did: did.String(), - }) -} - -// GET /xrpc/com.atproto.identity.resolveDid -func (srv *Server) ResolveDid(c echo.Context) error { - ctx := c.Request().Context() - - did, err := syntax.ParseDID(c.QueryParam("did")) - if err != nil { - return c.JSON(400, GenericError{ - Error: "InvalidDidSyntax", - Message: err.Error(), - }) - } - - rawDoc, err := srv.dir.ResolveDIDRaw(ctx, did) - if err != nil && errors.Is(err, identity.ErrDIDNotFound) { - return c.JSON(404, GenericError{ - Error: "DidNotFound", - Message: err.Error(), - }) - } else if err != nil { - return c.JSON(500, GenericError{ - Error: "InternalError", - Message: err.Error(), - }) - } - return c.JSON(200, comatproto.IdentityResolveDid_Output{ - DidDoc: rawDoc, - }) -} - -// helper for resolveIdentity -func (srv *Server) resolveIdentityFromHandle(c echo.Context, handle syntax.Handle) error { - ctx := c.Request().Context() - - did, err := srv.dir.ResolveHandle(ctx, handle) - if err != nil && errors.Is(err, identity.ErrHandleNotFound) { - return c.JSON(404, GenericError{ - Error: "HandleNotFound", - Message: err.Error(), - }) - } else if err != nil { - srv.logger.Warn("failed handle resolution", "err", err, "handle", handle) - return c.JSON(502, GenericError{ - Error: "HandleResolutionFailed", - Message: err.Error(), - }) - } - - rawDoc, err := srv.dir.ResolveDIDRaw(ctx, did) - if err != nil && errors.Is(err, identity.ErrDIDNotFound) { - return c.JSON(404, GenericError{ - Error: "DidNotFound", - Message: err.Error(), - }) - } else if err != nil { - return c.JSON(502, GenericError{ - Error: "DIDResolutionFailed", - Message: err.Error(), - }) - } - - var doc identity.DIDDocument - if err := json.Unmarshal(rawDoc, &doc); err != nil { - return c.JSON(400, GenericError{ - Error: "InvalidDidDocument", - Message: err.Error(), - }) - } - - ident := identity.ParseIdentity(&doc) - declHandle, err := ident.DeclaredHandle() - if err != nil || declHandle != handle { - return c.JSON(400, GenericError{ - Error: "HandleMismatch", - Message: err.Error(), - }) - } - - return c.JSON(200, comatproto.IdentityDefs_IdentityInfo{ - Did: ident.DID.String(), - Handle: handle.String(), - DidDoc: rawDoc, - }) -} - -// helper for resolveIdentity -func (srv *Server) resolveIdentityFromDID(c echo.Context, did syntax.DID) error { - ctx := c.Request().Context() - - rawDoc, err := srv.dir.ResolveDIDRaw(ctx, did) - if err != nil && errors.Is(err, identity.ErrDIDNotFound) { - return c.JSON(404, GenericError{ - Error: "DidNotFound", - Message: err.Error(), - }) - } else if err != nil { - return c.JSON(502, GenericError{ - Error: "DIDResolutionFailed", - Message: err.Error(), - }) - } - - var doc identity.DIDDocument - if err := json.Unmarshal(rawDoc, &doc); err != nil { - return c.JSON(400, GenericError{ - Error: "InvalidDidDocument", - Message: err.Error(), - }) - } - - ident := identity.ParseIdentity(&doc) - handle, err := ident.DeclaredHandle() - if err != nil { - // no handle declared, or invalid syntax - handle = syntax.Handle("handle.invalid") - } - - checkDID, err := srv.dir.ResolveHandle(ctx, handle) - if err != nil || checkDID != did { - handle = syntax.Handle("handle.invalid") - } - - return c.JSON(200, comatproto.IdentityDefs_IdentityInfo{ - Did: ident.DID.String(), - Handle: handle.String(), - DidDoc: rawDoc, - }) -} - -// GET /xrpc/com.atproto.identity.resolveIdentity -func (srv *Server) ResolveIdentity(c echo.Context) error { - // we partially re-implement the "Lookup()" logic here, but returning the full DID document, not `identity.Identity` - atid, err := syntax.ParseAtIdentifier(c.QueryParam("identifier")) - if err != nil { - return c.JSON(400, GenericError{ - Error: "InvalidIdentifierSyntax", - Message: err.Error(), - }) - } - - handle, err := atid.AsHandle() - if nil == err { - return srv.resolveIdentityFromHandle(c, handle) - } - did, err := atid.AsDID() - if nil == err { - return srv.resolveIdentityFromDID(c, did) - } - return fmt.Errorf("unreachable code path") -} - -// POST /xrpc/com.atproto.identity.refreshIdentity -func (srv *Server) RefreshIdentity(c echo.Context) error { - ctx := c.Request().Context() - - var body comatproto.IdentityRefreshIdentity_Input - if err := c.Bind(&body); err != nil { - return c.JSON(400, GenericError{ - Error: "InvalidRequestBody", - Message: err.Error(), - }) - } - - atid, err := syntax.ParseAtIdentifier(body.Identifier) - if err != nil { - return c.JSON(400, GenericError{ - Error: "InvalidIdentifierSyntax", - Message: err.Error(), - }) - } - - did, err := atid.AsDID() - if nil == err { - if err := srv.dir.PurgeDID(ctx, did); err != nil { - return err - } - return srv.resolveIdentityFromDID(c, did) - } - handle, err := atid.AsHandle() - if nil == err { - if err := srv.dir.PurgeHandle(ctx, handle); err != nil { - return err - } - return srv.resolveIdentityFromHandle(c, handle) - } - - return fmt.Errorf("unreachable code path") -} - -type GenericStatus struct { - Daemon string `json:"daemon"` - Status string `json:"status"` - Message string `json:"msg,omitempty"` -} - -func (s *Server) HandleHealthCheck(c echo.Context) error { - return c.JSON(200, GenericStatus{Status: "ok", Daemon: "domesday"}) -} - -func (srv *Server) WebHome(c echo.Context) error { - return c.String(200, ` -######## ####### ## ## ######## ###### ######## ### ## ## -## ## ## ## ### ### ## ## ## ## ## ## ## ## ## -## ## ## ## #### #### ## ## ## ## ## ## #### -## ## ## ## ## ### ## ###### ###### ## ## ## ## ## -## ## ## ## ## ## ## ## ## ## ######### ## -## ## ## ## ## ## ## ## ## ## ## ## ## ## -######## ####### ## ## ######## ###### ######## ## ## ## - -This is an AT Protocol Identity Service - -Most API routes are under /xrpc/ - - Code: https://github.com/bluesky-social/indigo/tree/main/cmd/domesday - Protocol: https://atproto.com - `) - -} diff --git a/cmd/domesday/main.go b/cmd/domesday/main.go deleted file mode 100644 --- a/cmd/domesday/main.go +++ /dev/null @@ -1,340 +0,0 @@ -package main - -import ( - "context" - "encoding/json" - "fmt" - "io" - "log/slog" - _ "net/http/pprof" - "os" - "runtime" - "strings" - - "github.com/bluesky-social/indigo/atproto/identity/apidir" - "github.com/bluesky-social/indigo/atproto/syntax" - - "github.com/carlmjohnson/versioninfo" - _ "github.com/joho/godotenv/autoload" - "github.com/urfave/cli/v2" -) - -func main() { - if err := run(os.Args); err != nil { - slog.Error("exiting", "err", err) - os.Exit(-1) - } -} - -func run(args []string) error { - - app := cli.App{ - Name: "domesday", - Usage: "atproto identity directory", - Version: versioninfo.Short(), - Flags: []cli.Flag{ - &cli.StringFlag{ - Name: "atp-relay-host", - Usage: "hostname and port of Relay to subscribe to", - Value: "wss://bsky.network", - EnvVars: []string{"ATP_RELAY_HOST", "ATP_BGS_HOST"}, - }, - &cli.StringFlag{ - Name: "atp-plc-host", - Usage: "method, hostname, and port of PLC registry", - Value: "https://plc.directory", - EnvVars: []string{"ATP_PLC_HOST"}, - }, - &cli.IntFlag{ - Name: "plc-rate-limit", - Usage: "max number of requests per second to PLC registry", - Value: 300, - EnvVars: []string{"DOMESDAY_PLC_RATE_LIMIT"}, - }, - &cli.StringFlag{ - Name: "redis-url", - Usage: "redis connection URL: redis://:@:6379/", - Value: "redis://localhost:6379/0", - EnvVars: []string{"DOMESDAY_REDIS_URL"}, - }, - &cli.StringFlag{ - Name: "log-level", - Usage: "log verbosity level (eg: warn, info, debug)", - EnvVars: []string{"DOMESDAY_LOG_LEVEL", "GO_LOG_LEVEL", "LOG_LEVEL"}, - }, - }, - Commands: []*cli.Command{ - &cli.Command{ - Name: "serve", - Usage: "run the domesday API daemon", - Action: runServeCmd, - Flags: []cli.Flag{ - &cli.StringFlag{ - Name: "bind", - Usage: "Specify the local IP/port to bind to", - Required: false, - Value: ":6600", - EnvVars: []string{"DOMESDAY_BIND"}, - }, - &cli.StringFlag{ - Name: "metrics-listen", - Usage: "IP or address, and port, to listen on for metrics APIs", - Value: ":3989", - EnvVars: []string{"DOMESDAY_METRICS_LISTEN"}, - }, - &cli.BoolFlag{ - Name: "disable-firehose-consumer", - Usage: "don't consume #identity events from firehose", - EnvVars: []string{"DOMESDAY_DISABLE_FIREHOSE_CONSUMER"}, - }, - &cli.BoolFlag{ - Name: "disable-refresh", - Usage: "disable the refreshIdentity API endpoint", - EnvVars: []string{"DOMESDAY_DISABLE_REFRESH"}, - }, - &cli.IntFlag{ - Name: "firehose-parallelism", - Usage: "number of concurrent firehose workers", - Value: 4, - EnvVars: []string{"DOMESDAY_FIREHOSE_PARALLELISM"}, - }, - }, - }, - &cli.Command{ - Name: "resolve-handle", - ArgsUsage: ``, - Usage: "query service for handle resoltion", - Action: runResolveHandleCmd, - Flags: []cli.Flag{ - &cli.StringFlag{ - Name: "host", - Usage: "domesday server to send request to", - Value: "http://localhost:6600", - EnvVars: []string{"DOMESDAY_HOST"}, - }, - }, - }, - &cli.Command{ - Name: "resolve-did", - ArgsUsage: ``, - Usage: "query service for DID document resoltion", - Action: runResolveDIDCmd, - Flags: []cli.Flag{ - &cli.StringFlag{ - Name: "host", - Usage: "domesday server to send request to", - Value: "http://localhost:6600", - EnvVars: []string{"DOMESDAY_HOST"}, - }, - }, - }, - &cli.Command{ - Name: "lookup", - ArgsUsage: ``, - Usage: "query service for identity resoltion", - Action: runLookupCmd, - Flags: []cli.Flag{ - &cli.StringFlag{ - Name: "host", - Usage: "domesday server to send request to", - Value: "http://localhost:6600", - EnvVars: []string{"DOMESDAY_HOST"}, - }, - }, - }, - &cli.Command{ - Name: "refresh", - ArgsUsage: ``, - Usage: "ask service to refresh identity", - Action: runRefreshCmd, - Flags: []cli.Flag{ - &cli.StringFlag{ - Name: "host", - Usage: "domesday server to send request to", - Value: "http://localhost:6600", - EnvVars: []string{"DOMESDAY_HOST"}, - }, - }, - }, - }, - } - - return app.Run(args) -} - -func configLogger(cctx *cli.Context, writer io.Writer) *slog.Logger { - var level slog.Level - switch strings.ToLower(cctx.String("log-level")) { - case "error": - level = slog.LevelError - case "warn": - level = slog.LevelWarn - case "info": - level = slog.LevelInfo - case "debug": - level = slog.LevelDebug - default: - level = slog.LevelInfo - } - logger := slog.New(slog.NewJSONHandler(writer, &slog.HandlerOptions{ - Level: level, - })) - slog.SetDefault(logger) - return logger -} - -func configClient(cctx *cli.Context) apidir.APIDirectory { - return apidir.NewAPIDirectory(cctx.String("host")) -} - -func runServeCmd(cctx *cli.Context) error { - logger := configLogger(cctx, os.Stdout) - ctx := context.Background() - - srv, err := NewServer( - Config{ - Logger: logger, - Bind: cctx.String("bind"), - RedisURL: cctx.String("redis-url"), - PLCHost: cctx.String("atp-plc-host"), - PLCRateLimit: cctx.Int("plc-rate-limit"), - DisableRefresh: cctx.Bool("disable-refresh"), - }, - ) - if err != nil { - return fmt.Errorf("failed to construct server: %v", err) - } - - if !cctx.Bool("disable-firehose-consumer") { - go func() { - firehoseHost := cctx.String("atp-relay-host") - firehoseParallelism := cctx.Int("firehose-parallelism") - if err := srv.RunFirehoseConsumer(ctx, firehoseHost, firehoseParallelism); err != nil { - slog.Error("firehose consumer thread failed", "err", err) - // NOTE: not crashing or halting process here - } - }() - go func() { - if err := srv.RunPersistCursor(ctx); err != nil { - slog.Error("firehose persist thread failed", "err", err) - // NOTE: not crashing or halting process here - } - }() - } - - // prometheus HTTP endpoint: /metrics - go func() { - // TODO: what is this tuning for? just cargo-culted it - runtime.SetBlockProfileRate(10) - runtime.SetMutexProfileFraction(10) - if err := srv.RunMetrics(cctx.String("metrics-listen")); err != nil { - slog.Error("failed to start metrics endpoint", "error", err) - // NOTE: not crashing or halting process here - } - }() - - return srv.RunAPI() -} - -func runResolveHandleCmd(cctx *cli.Context) error { - ctx := context.Background() - dir := configClient(cctx) - - s := cctx.Args().First() - if s == "" { - return fmt.Errorf("need to provide identifier for resolution") - } - handle, err := syntax.ParseHandle(s) - if err != nil { - return err - } - - did, err := dir.ResolveHandle(ctx, handle) - if err != nil { - return err - } - fmt.Println(did.String()) - return nil -} - -func runResolveDIDCmd(cctx *cli.Context) error { - ctx := context.Background() - dir := configClient(cctx) - - s := cctx.Args().First() - if s == "" { - return fmt.Errorf("need to provide identifier for resolution") - } - did, err := syntax.ParseDID(s) - if err != nil { - return err - } - - raw, err := dir.ResolveDIDRaw(ctx, did) - if err != nil { - return err - } - b, err := json.MarshalIndent(raw, "", " ") - if err != nil { - return err - } - fmt.Println(string(b)) - return nil -} - -func runLookupCmd(cctx *cli.Context) error { - ctx := context.Background() - dir := configClient(cctx) - - s := cctx.Args().First() - if s == "" { - return fmt.Errorf("need to provide identifier for resolution") - } - atid, err := syntax.ParseAtIdentifier(s) - if err != nil { - return err - } - - ident, err := dir.Lookup(ctx, *atid) - if err != nil { - return err - } - - b, err := json.MarshalIndent(ident, "", " ") - if err != nil { - return err - } - fmt.Println(string(b)) - return nil -} - -func runRefreshCmd(cctx *cli.Context) error { - ctx := context.Background() - dir := configClient(cctx) - - s := cctx.Args().First() - if s == "" { - return fmt.Errorf("need to provide identifier for resolution") - } - atid, err := syntax.ParseAtIdentifier(s) - if err != nil { - return err - } - - err = dir.Purge(ctx, *atid) - if err != nil { - return err - } - - ident, err := dir.Lookup(ctx, *atid) - if err != nil { - return err - } - - b, err := json.MarshalIndent(ident, "", " ") - if err != nil { - return err - } - fmt.Println(string(b)) - return nil -} diff --git a/cmd/domesday/metrics.go b/cmd/domesday/metrics.go deleted file mode 100644 --- a/cmd/domesday/metrics.go +++ /dev/null @@ -1,58 +0,0 @@ -package main - -import ( - "github.com/prometheus/client_golang/prometheus" - "github.com/prometheus/client_golang/prometheus/promauto" -) - -var handleCacheHits = promauto.NewCounter(prometheus.CounterOpts{ - Name: "domesday_resolve_handle_cache_hits", - Help: "Number of cache hits for ATProto handle resolutions", -}) - -var handleCacheMisses = promauto.NewCounter(prometheus.CounterOpts{ - Name: "domesday_resolve_handle_cache_misses", - Help: "Number of cache misses for ATProto handle resolutions", -}) - -var handleRequestsCoalesced = promauto.NewCounter(prometheus.CounterOpts{ - Name: "domesday_resolve_handle_requests_coalesced", - Help: "Number of handle requests coalesced", -}) - -var handleResolutionErrors = promauto.NewCounter(prometheus.CounterOpts{ - Name: "domesday_resolve_handle_resolution_errors", - Help: "Number of non-cached handle resolution errors", -}) - -var handleResolveDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ - Name: "domesday_resolve_handle_duration", - Help: "Time to resolve a handle from network (not cached)", - Buckets: prometheus.ExponentialBucketsRange(0.001, 2, 15), -}, []string{"status"}) - -var didCacheHits = promauto.NewCounter(prometheus.CounterOpts{ - Name: "domesday_resolve_did_cache_hits", - Help: "Number of cache hits for ATProto DID resolutions", -}) - -var didCacheMisses = promauto.NewCounter(prometheus.CounterOpts{ - Name: "domesday_resolve_did_cache_misses", - Help: "Number of cache misses for ATProto DID resolutions", -}) - -var didRequestsCoalesced = promauto.NewCounter(prometheus.CounterOpts{ - Name: "domesday_resolve_did_requests_coalesced", - Help: "Number of DID requests coalesced", -}) - -var didResolutionErrors = promauto.NewCounter(prometheus.CounterOpts{ - Name: "domesday_resolve_did_resolution_errors", - Help: "Number of non-cached DID resolution errors", -}) - -var didResolveDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ - Name: "domesday_resolve_did_duration", - Help: "Time to resolve a DID from network (not cached)", - Buckets: prometheus.ExponentialBucketsRange(0.001, 2, 15), -}, []string{"status"}) diff --git a/cmd/domesday/resolver.go b/cmd/domesday/resolver.go deleted file mode 100644 --- a/cmd/domesday/resolver.go +++ /dev/null @@ -1,319 +0,0 @@ -package main - -import ( - "context" - "encoding/json" - "errors" - "fmt" - "log/slog" - "sync" - "time" - - "github.com/bluesky-social/indigo/atproto/identity" - "github.com/bluesky-social/indigo/atproto/syntax" - - "github.com/go-redis/cache/v9" - "github.com/redis/go-redis/v9" -) - -// This file is a fork of indigo:atproto/identity/redisdir. It stores raw DID documents, not identities, and implements `identity.Resolver`. - -// Uses redis as a cache for identity lookups. -// -// Includes an in-process LRU cache as well (provided by the redis client library), for hot key (identities). -type RedisResolver struct { - Inner identity.Resolver - ErrTTL time.Duration - HitTTL time.Duration - InvalidHandleTTL time.Duration - Logger *slog.Logger - - handleCache *cache.Cache - didCache *cache.Cache - didResolveChans sync.Map - handleResolveChans sync.Map -} - -type handleEntry struct { - Updated time.Time - // needs to be pointer type, because unmarshalling empty string would be an error - DID *syntax.DID - Err error -} - -type didEntry struct { - Updated time.Time - RawDoc json.RawMessage - Err error -} - -var _ identity.Resolver = (*RedisResolver)(nil) - -// Creates a new caching `identity.Resolver` wrapper around an existing directory, using Redis and in-process LRU for caching. -// -// `redisURL` contains all the redis connection config options. -// `hitTTL` and `errTTL` define how long successful and errored identity metadata should be cached (respectively). errTTL is expected to be shorted than hitTTL. -// `lruSize` is the size of the in-process cache, for each of the handle and identity caches. 10000 is a reasonable default. -// -// NOTE: Errors returned may be inconsistent with the base directory, or between calls. This is because cached errors are serialized/deserialized and that may break equality checks. -func NewRedisResolver(inner identity.Resolver, redisURL string, hitTTL, errTTL, invalidHandleTTL time.Duration, lruSize int) (*RedisResolver, error) { - opt, err := redis.ParseURL(redisURL) - if err != nil { - return nil, fmt.Errorf("could not configure redis identity cache: %w", err) - } - rdb := redis.NewClient(opt) - // check redis connection - _, err = rdb.Ping(context.TODO()).Result() - if err != nil { - return nil, fmt.Errorf("could not connect to redis identity cache: %w", err) - } - handleCache := cache.New(&cache.Options{ - Redis: rdb, - LocalCache: cache.NewTinyLFU(lruSize, hitTTL), - }) - didCache := cache.New(&cache.Options{ - Redis: rdb, - LocalCache: cache.NewTinyLFU(lruSize, hitTTL), - }) - return &RedisResolver{ - Inner: inner, - ErrTTL: errTTL, - HitTTL: hitTTL, - InvalidHandleTTL: invalidHandleTTL, - handleCache: handleCache, - didCache: didCache, - }, nil -} - -func (d *RedisResolver) isHandleStale(e *handleEntry) bool { - if e.Err != nil && time.Since(e.Updated) > d.ErrTTL { - return true - } - return false -} - -func (d *RedisResolver) isDIDStale(e *didEntry) bool { - if e.Err != nil && time.Since(e.Updated) > d.ErrTTL { - return true - } - return false -} - -func (d *RedisResolver) refreshHandle(ctx context.Context, h syntax.Handle) handleEntry { - start := time.Now() - did, err := d.Inner.ResolveHandle(ctx, h) - duration := time.Since(start) - - if err != nil { - d.Logger.Info("handle resolution failed", "handle", h, "duration", duration, "err", err) - handleResolutionErrors.Inc() - handleResolveDuration.WithLabelValues("fail").Observe(time.Since(start).Seconds()) - } else { - handleResolveDuration.WithLabelValues("success").Observe(time.Since(start).Seconds()) - } - if duration.Seconds() > 5.0 { - d.Logger.Info("slow handle resolution", "handle", h, "duration", duration) - } - - he := handleEntry{ - Updated: time.Now(), - DID: &did, - Err: err, - } - err = d.handleCache.Set(&cache.Item{ - Ctx: ctx, - Key: "domes/handle/" + h.String(), - Value: he, - TTL: d.ErrTTL, - }) - if err != nil { - d.Logger.Error("identity cache write failed", "cache", "handle", "err", err) - } - return he -} - -func (d *RedisResolver) refreshDID(ctx context.Context, did syntax.DID) didEntry { - start := time.Now() - rawDoc, err := d.Inner.ResolveDIDRaw(ctx, did) - duration := time.Since(start) - - if err != nil { - d.Logger.Info("DID resolution failed", "did", did, "duration", duration, "err", err) - didResolutionErrors.Inc() - didResolveDuration.WithLabelValues("fail").Observe(time.Since(start).Seconds()) - } else { - didResolveDuration.WithLabelValues("success").Observe(time.Since(start).Seconds()) - } - if duration.Seconds() > 5.0 { - d.Logger.Info("slow DID resolution", "did", did, "duration", duration) - } - - // persist the DID lookup error, instead of processing it immediately - entry := didEntry{ - Updated: time.Now(), - RawDoc: rawDoc, - Err: err, - } - - err = d.didCache.Set(&cache.Item{ - Ctx: ctx, - Key: "domes/did/" + did.String(), - Value: entry, - TTL: d.HitTTL, - }) - if err != nil { - d.Logger.Error("DID cache write failed", "cache", "did", "did", did, "err", err) - } - return entry -} - -func (d *RedisResolver) ResolveHandle(ctx context.Context, h syntax.Handle) (syntax.DID, error) { - if h.IsInvalidHandle() { - return "", fmt.Errorf("can not resolve handle: %w", identity.ErrInvalidHandle) - } - h = h.Normalize() - var entry handleEntry - err := d.handleCache.Get(ctx, "domes/handle/"+h.String(), &entry) - if err != nil && err != cache.ErrCacheMiss { - return "", fmt.Errorf("identity cache read failed: %w", err) - } - if err == nil && !d.isHandleStale(&entry) { // if no error... - handleCacheHits.Inc() - if entry.Err != nil { - return "", entry.Err - } else if entry.DID != nil { - return *entry.DID, nil - } else { - return "", errors.New("code flow error in redis identity directory") - } - } - handleCacheMisses.Inc() - - // Coalesce multiple requests for the same Handle - res := make(chan struct{}) - val, loaded := d.handleResolveChans.LoadOrStore(h.String(), res) - if loaded { - handleRequestsCoalesced.Inc() - // Wait for the result from the pending request - select { - case <-val.(chan struct{}): - // The result should now be in the cache - err := d.handleCache.Get(ctx, "domes/handle/"+h.String(), entry) - if err != nil && err != cache.ErrCacheMiss { - return "", fmt.Errorf("identity cache read failed: %w", err) - } - if err == nil && !d.isHandleStale(&entry) { // if no error... - if entry.Err != nil { - return "", entry.Err - } else if entry.DID != nil { - return *entry.DID, nil - } else { - return "", errors.New("code flow error in redis identity directory") - } - } - return "", errors.New("identity not found in cache after coalesce returned") - case <-ctx.Done(): - return "", ctx.Err() - } - } - - // Update the Handle Entry from PLC and cache the result - newEntry := d.refreshHandle(ctx, h) - - // Cleanup the coalesce map and close the results channel - d.handleResolveChans.Delete(h.String()) - // Callers waiting will now get the result from the cache - close(res) - - if newEntry.Err != nil { - return "", newEntry.Err - } - if newEntry.DID != nil { - return *newEntry.DID, nil - } - return "", errors.New("unexpected control-flow error") -} - -func (d *RedisResolver) ResolveDIDRaw(ctx context.Context, did syntax.DID) (json.RawMessage, error) { - var entry didEntry - err := d.didCache.Get(ctx, "domes/did/"+did.String(), &entry) - if err != nil && err != cache.ErrCacheMiss { - return nil, fmt.Errorf("DID cache read failed: %w", err) - } - if err == nil && !d.isDIDStale(&entry) { // if no error... - didCacheHits.Inc() - return entry.RawDoc, entry.Err - } - didCacheMisses.Inc() - - // Coalesce multiple requests for the same DID - res := make(chan struct{}) - val, loaded := d.didResolveChans.LoadOrStore(did.String(), res) - if loaded { - didRequestsCoalesced.Inc() - // Wait for the result from the pending request - select { - case <-val.(chan struct{}): - // The result should now be in the cache - err = d.didCache.Get(ctx, "domes/did/"+did.String(), &entry) - if err != nil && err != cache.ErrCacheMiss { - return nil, fmt.Errorf("DID cache read failed: %w", err) - } - if err == nil && !d.isDIDStale(&entry) { // if no error... - return entry.RawDoc, entry.Err - } - return nil, errors.New("DID not found in cache after coalesce returned") - case <-ctx.Done(): - return nil, ctx.Err() - } - } - - // Update the DID Entry and cache the result - newEntry := d.refreshDID(ctx, did) - - // Cleanup the coalesce map and close the results channel - d.didResolveChans.Delete(did.String()) - // Callers waiting will now get the result from the cache - close(res) - - if newEntry.Err != nil { - return nil, newEntry.Err - } - if newEntry.RawDoc != nil { - return newEntry.RawDoc, nil - } - return nil, errors.New("unexpected control-flow error") -} - -func (d *RedisResolver) ResolveDID(ctx context.Context, did syntax.DID) (*identity.DIDDocument, error) { - b, err := d.ResolveDIDRaw(ctx, did) - if err != nil { - return nil, err - } - - var doc identity.DIDDocument - if err := json.Unmarshal(b, &doc); err != nil { - return nil, fmt.Errorf("%w: JSON DID document parse: %w", identity.ErrDIDResolutionFailed, err) - } - if doc.DID != did { - return nil, fmt.Errorf("document ID did not match DID") - } - return &doc, nil -} - -func (d *RedisResolver) PurgeHandle(ctx context.Context, handle syntax.Handle) error { - handle = handle.Normalize() - err := d.handleCache.Delete(ctx, "domes/handle/"+handle.String()) - if err == cache.ErrCacheMiss { - return nil - } - return err -} - -func (d *RedisResolver) PurgeDID(ctx context.Context, did syntax.DID) error { - err := d.didCache.Delete(ctx, "domes/did/"+did.String()) - if err == cache.ErrCacheMiss { - return nil - } - return err -} diff --git a/cmd/domesday/server.go b/cmd/domesday/server.go deleted file mode 100644 --- a/cmd/domesday/server.go +++ /dev/null @@ -1,218 +0,0 @@ -package main - -import ( - "context" - "errors" - "fmt" - "log/slog" - "net" - "net/http" - "os" - "os/signal" - "syscall" - "time" - - "github.com/bluesky-social/indigo/atproto/identity" - - "github.com/labstack/echo/v4" - "github.com/labstack/echo/v4/middleware" - "github.com/prometheus/client_golang/prometheus/promhttp" - "github.com/redis/go-redis/v9" - slogecho "github.com/samber/slog-echo" - "golang.org/x/time/rate" -) - -type Server struct { - dir *RedisResolver - echo *echo.Echo - httpd *http.Server - logger *slog.Logger - - // this redis client is used to store firehose offset - redisClient *redis.Client - - // lastSeq is the most recent event sequence number we've received and begun to handle. - // This number is periodically persisted to redis, if redis is present. - // The value is best-effort (the stream handling itself is concurrent, so event numbers may not be monotonic), - // but nonetheless, you must use atomics when updating or reading this (to avoid data races). - lastSeq int64 -} - -type Config struct { - Logger *slog.Logger - PLCHost string - PLCRateLimit int - RedisURL string - Bind string - DisableRefresh bool -} - -func NewServer(config Config) (*Server, error) { - logger := config.Logger - if logger == nil { - logger = slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{ - Level: slog.LevelInfo, - })) - } - - baseDir := identity.BaseDirectory{ - PLCURL: config.PLCHost, - HTTPClient: http.Client{ - Timeout: time.Second * 10, - Transport: &http.Transport{ - // would want this around 100ms for services doing lots of handle resolution (to reduce number of idle connections). Impacts PLC connections as well, but not too bad. - IdleConnTimeout: time.Millisecond * 100, - MaxIdleConns: 1000, - }, - }, - Resolver: net.Resolver{ - Dial: func(ctx context.Context, network, address string) (net.Conn, error) { - d := net.Dialer{Timeout: time.Second * 3} - return d.DialContext(ctx, network, address) - }, - }, - PLCLimiter: rate.NewLimiter(rate.Limit(config.PLCRateLimit), 1), - TryAuthoritativeDNS: true, - SkipDNSDomainSuffixes: []string{".bsky.social", ".staging.bsky.dev"}, - // TODO: UserAgent: "domesday", - } - - // TODO: config these timeouts - redisDir, err := NewRedisResolver(&baseDir, config.RedisURL, time.Hour*24, time.Minute*2, time.Minute*5, 50_000) - if err != nil { - return nil, err - } - redisDir.Logger = logger - - // configure redis client (for firehose consumer) - redisOpt, err := redis.ParseURL(config.RedisURL) - if err != nil { - return nil, fmt.Errorf("parsing redis URL: %v", err) - } - redisClient := redis.NewClient(redisOpt) - // check redis connection - _, err = redisClient.Ping(context.Background()).Result() - if err != nil { - return nil, fmt.Errorf("redis ping failed: %v", err) - } - - e := echo.New() - - // httpd - var ( - httpTimeout = 1 * time.Minute - httpMaxHeaderBytes = 1 * (1024 * 1024) - ) - - srv := &Server{ - echo: e, - dir: redisDir, - logger: logger, - redisClient: redisClient, - } - - srv.httpd = &http.Server{ - Handler: srv, - Addr: config.Bind, - WriteTimeout: httpTimeout, - ReadTimeout: httpTimeout, - MaxHeaderBytes: httpMaxHeaderBytes, - } - - e.HideBanner = true - e.Use(slogecho.New(logger)) - e.Use(middleware.Recover()) - e.Use(middleware.BodyLimit("4M")) - e.HTTPErrorHandler = srv.errorHandler - e.Use(middleware.SecureWithConfig(middleware.SecureConfig{ - ContentTypeNosniff: "nosniff", - XFrameOptions: "SAMEORIGIN", - HSTSMaxAge: 31536000, // 365 days - // TODO: - // ContentSecurityPolicy - // XSSProtection - })) - - e.GET("/", srv.WebHome) - e.GET("/_health", srv.HandleHealthCheck) - e.GET("/xrpc/com.atproto.identity.resolveHandle", srv.ResolveHandle) - e.GET("/xrpc/com.atproto.identity.resolveDid", srv.ResolveDid) - e.GET("/xrpc/com.atproto.identity.resolveIdentity", srv.ResolveIdentity) - if !config.DisableRefresh { - e.POST("/xrpc/com.atproto.identity.refreshIdentity", srv.RefreshIdentity) - } - - return srv, nil -} - -func (srv *Server) ServeHTTP(rw http.ResponseWriter, req *http.Request) { - srv.echo.ServeHTTP(rw, req) -} - -func (srv *Server) RunAPI() error { - srv.logger.Info("starting server", "bind", srv.httpd.Addr) - go func() { - if err := srv.httpd.ListenAndServe(); err != nil { - if !errors.Is(err, http.ErrServerClosed) { - srv.logger.Error("HTTP server shutting down unexpectedly", "err", err) - } - } - }() - - // Wait for a signal to exit. - srv.logger.Info("registering OS exit signal handler") - quit := make(chan struct{}) - exitSignals := make(chan os.Signal, 1) - signal.Notify(exitSignals, syscall.SIGINT, syscall.SIGTERM) - go func() { - sig := <-exitSignals - srv.logger.Info("received OS exit signal", "signal", sig) - - // Shut down the HTTP server - if err := srv.Shutdown(); err != nil { - srv.logger.Error("HTTP server shutdown error", "err", err) - } - - // Trigger the return that causes an exit. - close(quit) - }() - <-quit - srv.logger.Info("graceful shutdown complete") - return nil -} - -func (srv *Server) RunMetrics(bind string) error { - p := "/metrics" - srv.logger.Info("starting metrics endpoint", "bind", bind, "path", p) - http.Handle(p, promhttp.Handler()) - return http.ListenAndServe(bind, nil) -} - -func (srv *Server) Shutdown() error { - srv.logger.Info("shutting down") - - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - - return srv.httpd.Shutdown(ctx) -} - -type GenericError struct { - Error string `json:"error"` - Message string `json:"message"` -} - -func (srv *Server) errorHandler(err error, c echo.Context) { - code := http.StatusInternalServerError - var errorMessage string - if he, ok := err.(*echo.HTTPError); ok { - code = he.Code - errorMessage = fmt.Sprintf("%s", he.Message) - } - if code >= 500 { - srv.logger.Warn("domesday-http-internal-error", "err", err) - } - if !c.Response().Committed { - c.JSON(code, GenericError{Error: "InternalError", Message: errorMessage}) - } -}