Add Tap mode (#2)

* tap mode

* add acks

* update dockercompose with tap

* update readme

README.md

···
 A small service that receives events from the AT firehose and produces them to Kafka. Supports standard JSON outputs as well as [Osprey](https://github.com/roostorg/osprey)
 formatted events.
 
+Additionally, at-kafka supports subscribing to [Tap](https://github.com/bluesky-social/indigo/tree/main/cmd/tap) if youare attempting to perform a network backfill.
+
 ## Usage
 
 ### Docker Compose
···
 
 ```yaml
 environment:
-  ATKAFKA_RELAY_HOST: "wss://bsky.network"
-  ATKAFKA_BOOTSTRAP_SERVERS: "kafka:29092"
-  ATKAFKA_OUTPUT_TOPIC: "atproto-events"
-  ATKAFKA_OSPREY_COMPATIBLE: "false"
-```
+  # For relay mode
+  ATKAFKA_RELAY_HOST: "wss://bsky.network" # ATProto relay to subscribe to for events
 
-Then start:
+  # For tap mode
+  ATKAFKA_TAP_HOST: "ws://localhost:2480" # Tap websocket host to subscribe to for events
+  ATKAFKA_DISABLE_ACKS: false # Whether to disable sending of acks to Tap
 
-```bash
-docker compose up -d        # Start services
-```
-
-### Docker Run
+  # Kafka configuration
+  ATKAFKA_BOOTSTRAP_SERVERS: "kafka:29092" # Kafka bootstrap servers, comma separated
+  ATKAFKA_OUTPUT_TOPIC: "atproto-events" # The output topic for events
+  ATKAFKA_OSPREY_COMPATIBLE: false # Whether to produce Osprey-compatible events
 
-For standard mode:
+  # Match only Blacksky PDS users
+  ATKAFKA_MATCHED_SERVICES: "blacksky.app" # A comma-separated list of PDSes to emit events for
+  # OR ignore anyone on Bluesky PBC PDSes
+  ATKAFKA_IGNORED_SERVICES: "*.bsky.network" # OR a comma-separated list of PDSes to _not_ emit events for
 
-```bash
-docker run -d \
-  -e ATKAFKA_BOOTSTRAP_SERVERS=kafka:9092 \
-  -e ATKAFKA_OUTPUT_TOPIC=atproto-events \
-  -p 2112:2112 \
-  ghcr.io/haileyok/at-kafka:latest
+  # Match only Teal.fm records
+  ATKAFKA_MATCHED_COLLECTIONS: "fm.teal.*" # A comma-separated list of collections to emit events for
+  # OR ignore all Bluesky records
+  ATKAFKA_IGNORED_COLLECTIONS: "app.bsky.*" # OR a comma-separated list of collections to ignore events for
 ```
 
-For Osprey-compatible mode:
+Then start:
 
 ```bash
-docker run -d \
-  -e ATKAFKA_BOOTSTRAP_SERVERS=kafka:9092 \
-  -e ATKAFKA_OUTPUT_TOPIC=atproto-events \
-  -e ATKAFKA_OSPREY_COMPATIBLE=true \
-  -p 2112:2112 \
-  ghcr.io/haileyok/at-kafka:latest
-```
+# For normal mode
+docker compose up -d
 
-## Configuration
+# For tap mode
+docker compose -f docker-compose.tap.yml up -d
 
-| Flag | Environment Variable | Default | Description |
-|------|---------------------|---------|-------------|
-| `--relay-host` | `ATKAFKA_RELAY_HOST` | `wss://bsky.network` | AT Protocol relay host to connect to |
-| `--bootstrap-servers` | `ATKAFKA_BOOTSTRAP_SERVERS` | (required) | Comma-separated list of Kafka bootstrap servers |
-| `--output-topic` | `ATKAFKA_OUTPUT_TOPIC` | (required) | Kafka topic to publish events to |
-| `--osprey-compatible` | `ATKAFKA_OSPREY_COMPATIBLE` | `false` | Enable Osprey-compatible event format |
+```
 
 ## Event Structure
 
···
 
 - `atkafka_handled_events` - Total events that are received on the firehose and handled
 - `atkafka_produced_events` - Total messages that are output on the bus
+- `atkafka_plc_requests` - Total number of PLC requests that were made, if applicable, and whether they were cached
+- `atkafka_api_requests` - Total number of API requests that were made, if applicable, and whether they were cached
+- `atkafka_cache_size` - The size of the PLC and API caches
+- `atkafka_acks_sent` - Total acks that were sent to Tap, if applicable

atkafka/atkafka.go

···
 
 type Server struct {
 	relayHost        string
+	tapHost          string
+	disableAcks      bool
 	bootstrapServers []string
 	outputTopic      string
 	ospreyCompat     bool
···
 	plcClient *PlcClient
 	apiClient *ApiClient
 	logger    *slog.Logger
+	ws        *websocket.Conn
 }
 
 type ServerArgs struct {
 	// network params
-	RelayHost string
-	PlcHost   string
-	ApiHost   string
+	RelayHost   string
+	TapHost     string
+	DisableAcks bool
+	PlcHost     string
+	ApiHost     string
 
 	// for watched and ignoed services or collections, only one list may be supplied
 	// for both services and collections, wildcards are acceptable. for example:
···
 
 	s := &Server{
 		relayHost:        args.RelayHost,
+		tapHost:          args.TapHost,
+		disableAcks:      args.DisableAcks,
 		plcClient:        plcClient,
 		apiClient:        apiClient,
 		bootstrapServers: args.BootstrapServers,

atkafka/metrics.go

···
 		Namespace: "atkafka",
 		Name:      "api_requests",
 	}, []string{"kind", "status", "cached"})
+
+	acksSent = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "atkafka",
+		Name:      "acks_sent",
+	}, []string{"status"})
 )

atkafka/tap.go

···
+package atkafka
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/url"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/araddon/dateparse"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+	"github.com/gorilla/websocket"
+	"github.com/twmb/franz-go/pkg/kgo"
+	"golang.org/x/sync/semaphore"
+)
+
+func (s *Server) RunTapMode(ctx context.Context) error {
+	sema := semaphore.NewWeighted(1_000)
+
+	s.logger.Info("starting tap consumer", "tap-host", s.tapHost, "bootstrap-servers", s.bootstrapServers, "output-topic", s.outputTopic)
+
+	createCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+
+	producerLogger := s.logger.With("component", "producer")
+	kafProducer, err := NewProducer(createCtx, producerLogger, s.bootstrapServers, s.outputTopic,
+		WithEnsureTopic(true),
+		WithTopicPartitions(200),
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create producer: %w", err)
+	}
+	defer kafProducer.Close()
+	s.producer = kafProducer
+	s.logger.Info("created producer")
+
+	wsDialer := websocket.DefaultDialer
+	u, err := url.Parse(s.tapHost)
+	if err != nil {
+		return fmt.Errorf("invalid tapHost: %w", err)
+	}
+	u.Path = "/channel"
+	s.logger.Info("created dialer")
+
+	wsErr := make(chan error, 1)
+	shutdownWs := make(chan struct{}, 1)
+	go func() {
+		logger := s.logger.With("component", "websocket")
+
+		logger.Info("subscribing to tap stream", "upstream", s.tapHost)
+
+		conn, _, err := wsDialer.Dial(u.String(), http.Header{
+			"User-Agent": []string{"at-kafka/0.0.0"},
+		})
+		if err != nil {
+			wsErr <- err
+			return
+		}
+		s.ws = conn
+
+		for {
+			var evt TapEvent
+			err := conn.ReadJSON(&evt)
+			if err != nil {
+				logger.Error("error reading json from websocket", "err", err)
+				break
+			}
+
+			if err := sema.Acquire(ctx, 1); err != nil {
+				logger.Error("error acquring sema", "err", err)
+				break
+			}
+
+			go func() {
+				defer sema.Release(1)
+				s.handleTapEvent(ctx, &evt)
+			}()
+		}
+
+		<-shutdownWs
+
+		wsErr <- nil
+	}()
+	s.logger.Info("created tap consumer")
+
+	signals := make(chan os.Signal, 1)
+	signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT)
+
+	select {
+	case sig := <-signals:
+		s.logger.Info("shutting down on signal", "signal", sig)
+	case err := <-wsErr:
+		if err != nil {
+			s.logger.Error("websocket error", "err", err)
+		} else {
+			s.logger.Info("websocket shutdown unexpectedly")
+		}
+	}
+
+	close(shutdownWs)
+
+	return nil
+}
+
+func (s *Server) handleTapEvent(ctx context.Context, evt *TapEvent) error {
+	logger := s.logger.With("component", "handleEvent")
+
+	var collection string
+	var actionName string
+
+	var evtKey string
+	var evtsToProduce [][]byte
+
+	if evt.Record != nil {
+		// key events by DID
+		evtKey = evt.Record.Did
+		did := evt.Record.Did
+		kind := evt.Record.Action
+		collection = evt.Record.Collection
+		rkey := evt.Record.Rkey
+		atUri := fmt.Sprintf("at://%s/%s/%s", did, collection, rkey)
+
+		skip := false
+		if len(s.watchedCollections) > 0 {
+			skip = true
+			for _, watchedCollection := range s.watchedCollections {
+				if watchedCollection == collection || strings.HasPrefix(collection, watchedCollection+".") {
+					skip = false
+					break
+				}
+			}
+		} else if len(s.ignoredCollections) > 0 {
+			for _, ignoredCollection := range s.ignoredCollections {
+				if ignoredCollection == collection || strings.HasPrefix(collection, ignoredCollection+".") {
+					skip = true
+					break
+				}
+			}
+		}
+
+		if skip {
+			logger.Debug("skipping event based on collection", "collection", collection)
+			return nil
+		}
+
+		actionName = "operation#" + kind
+
+		handledEvents.WithLabelValues(actionName, collection).Inc()
+
+		// create the formatted operation
+		atkOp := AtKafkaOp{
+			Action:     evt.Record.Action,
+			Collection: collection,
+			Rkey:       rkey,
+			Uri:        atUri,
+			Cid:        evt.Record.Cid,
+			Path:       fmt.Sprintf("%s/%s", collection, rkey),
+		}
+
+		if evt.Record.Record != nil {
+			atkOp.Record = *evt.Record.Record
+		}
+
+		kafkaEvt := AtKafkaEvent{
+			Did:       did,
+			Operation: &atkOp,
+		}
+
+		if evt.Record.Record != nil {
+			timestamp, err := parseTimeFromRecord(collection, *evt.Record.Record, rkey)
+			if err != nil {
+				return fmt.Errorf("error getting timestamp from record: %w", err)
+			}
+			kafkaEvt.Timestamp = timestamp.Format(time.RFC3339Nano)
+		}
+
+		evtBytes, err := json.Marshal(&kafkaEvt)
+		if err != nil {
+			return fmt.Errorf("failed to marshal kafka event: %w", err)
+		}
+
+		evtsToProduce = append(evtsToProduce, evtBytes)
+	}
+
+	for _, evtBytes := range evtsToProduce {
+		if err := s.produceAsyncTap(ctx, evtKey, evtBytes, evt.Id); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (s *Server) produceAsyncTap(ctx context.Context, key string, msg []byte, id int64) error {
+	logger := s.logger.With("name", "produceAsyncTap", "key", key, "id", id)
+	callback := func(r *kgo.Record, err error) {
+		status := "ok"
+		ackStatus := "ok"
+
+		go func() {
+			producedEvents.WithLabelValues(status).Inc()
+			if s.ws != nil && !s.disableAcks {
+				acksSent.WithLabelValues(ackStatus).Inc()
+			}
+		}()
+
+		if err != nil {
+			logger.Error("error producing message", "err", err)
+			status = "error"
+		} else if s.ws != nil && !s.disableAcks {
+			if err := s.ws.WriteJSON(TapAck{
+				Type: "ack",
+				Id:   id,
+			}); err != nil {
+				logger.Error("error sending ack", "err", err)
+				ackStatus = "error"
+			}
+		}
+	}
+
+	if err := s.producer.ProduceAsync(ctx, key, msg, callback); err != nil {
+		return fmt.Errorf("failed to produce message: %w", err)
+	}
+
+	return nil
+}
+
+func parseTimeFromRecord(collection string, rec map[string]any, rkey string) (*time.Time, error) {
+	var rkeyTime time.Time
+	if rkey != "self" {
+		rt, err := syntax.ParseTID(rkey)
+		if err == nil {
+			rkeyTime = rt.Time()
+		}
+	}
+
+	switch collection {
+	case "app.bsky.feed.post":
+		cat, ok := rec["createdAt"].(string)
+		if ok {
+			t, err := dateparse.ParseAny(cat)
+			if err == nil {
+				return &t, nil
+			}
+
+			if rkeyTime.IsZero() {
+				return timePtr(time.Now()), nil
+			}
+		}
+
+		return &rkeyTime, nil
+	case "app.bsky.feed.repost":
+		cat, ok := rec["createdAt"].(string)
+		if ok {
+			t, err := dateparse.ParseAny(cat)
+			if err == nil {
+				return &t, nil
+			}
+
+			if rkeyTime.IsZero() {
+				return nil, fmt.Errorf("failed to get a useful timestamp from record")
+			}
+		}
+
+		return &rkeyTime, nil
+	case "app.bsky.feed.like":
+		cat, ok := rec["createdAt"].(string)
+		if ok {
+			t, err := dateparse.ParseAny(cat)
+			if err == nil {
+				return &t, nil
+			}
+
+			if rkeyTime.IsZero() {
+				return nil, fmt.Errorf("failed to get a useful timestamp from record")
+			}
+		}
+
+		return &rkeyTime, nil
+	case "app.bsky.actor.profile":
+		// We can't really trust the createdat in the profile record anyway, and its very possible its missing. just use iat for this one
+		return timePtr(time.Now()), nil
+	case "app.bsky.feed.generator":
+		if !rkeyTime.IsZero() {
+			return &rkeyTime, nil
+		}
+		return timePtr(time.Now()), nil
+	default:
+		if !rkeyTime.IsZero() {
+			return &rkeyTime, nil
+		}
+		return timePtr(time.Now()), nil
+	}
+}
+
+func timePtr(t time.Time) *time.Time {
+	return &t
+}

atkafka/types.go

···
 	AccountAge   int64                               `json:"accountAge"`
 	Profile      *bsky.ActorDefs_ProfileViewDetailed `json:"profile"`
 }
+
+type TapEvent struct {
+	Id       int64             `json:"id"`
+	Type     string            `json:"type"`
+	Record   *TapEventRecord   `json:"record,omitempty"`
+	Identity *TapEventIdentity `json:"identity,omitempty"`
+}
+
+type TapEventRecord struct {
+	Live       bool            `json:"live"`
+	Rev        string          `json:"rev"`
+	Did        string          `json:"did"`
+	Collection string          `json:"collection"`
+	Rkey       string          `json:"rkey"`
+	Action     string          `json:"action"`
+	Cid        string          `json:"cid"`
+	Record     *map[string]any `json:"record,omitempty"`
+}
+
+type TapEventIdentity struct {
+	Did      string `json:"did"`
+	Handle   string `json:"handle"`
+	IsActive bool   `json:"isActive"`
+	Status   string `json:"status"`
+}
+
+type TapAck struct {
+	Type string `json:"type"`
+	Id   int64  `json:"id"`
+}

cmd/atkafka/main.go

···
 
 			return nil
 		},
+		Commands: cli.Commands{
+			&cli.Command{
+				Name: "tap-mode",
+				Flags: []cli.Flag{
+					&cli.StringFlag{
+						Name:    "tap-host",
+						Usage:   "Tap host to subscribe to for events",
+						Value:   "ws://localhost:2480",
+						EnvVars: []string{"ATKAFKA_TAP_HOST"},
+					},
+					&cli.BoolFlag{
+						Name:    "disable-acks",
+						Usage:   "Set to `true` to disable sending of event acks to Tap. May result in the loss of events.",
+						EnvVars: []string{"ATKAFKA_DISABLE_ACKS"},
+					},
+				},
+				Action: func(cmd *cli.Context) error {
+
+					ctx := context.Background()
+
+					telemetry.StartMetrics(cmd)
+					logger := telemetry.StartLogger(cmd)
+
+					s, err := atkafka.NewServer(&atkafka.ServerArgs{
+						TapHost:            cmd.String("tap-host"),
+						DisableAcks:        cmd.Bool("disable-acks"),
+						BootstrapServers:   cmd.StringSlice("bootstrap-servers"),
+						OutputTopic:        cmd.String("output-topic"),
+						WatchedCollections: cmd.StringSlice("watched-collections"),
+						IgnoredCollections: cmd.StringSlice("ignored-collections"),
+						Logger:             logger,
+					})
+					if err != nil {
+						return fmt.Errorf("failed to create new server: %w", err)
+					}
+
+					if err := s.RunTapMode(ctx); err != nil {
+						return fmt.Errorf("error running server: %w", err)
+					}
+
+					return nil
+				},
+			},
+		},
 	}
 
 	if err := app.Run(os.Args); err != nil {

docker-compose.tap.yml

···
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.6.0
+    hostname: zookeeper
+    container_name: zookeeper
+    ports:
+      - "2181:2181"
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    volumes:
+      - zookeeper-data:/var/lib/zookeeper/data
+      - zookeeper-logs:/var/lib/zookeeper/log
+
+  kafka:
+    image: confluentinc/cp-kafka:7.6.0
+    hostname: kafka
+    container_name: kafka
+    depends_on:
+      - zookeeper
+    ports:
+      - "9092:9092"
+      - "9101:9101"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_JMX_PORT: 9101
+      KAFKA_JMX_HOSTNAME: localhost
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
+    volumes:
+      - kafka-data:/var/lib/kafka/data
+
+
+  tap:
+    image: ghcr.io/bluesky-social/indigo/tap:latest
+    hostname: tap
+    container_name: tap
+    depends_on:
+      - kafka
+    ports:
+      - "2480:2480"
+    environment:
+      TAP_BIND: ":2480"
+      TAP_FULL_NETWORK: true
+      TAP_DISABLE_ACKS: false
+      TAP_COLLECTION_FILTERS: "app.bsky.graph.follow"
+    volumes:
+      - tap-data:/data
+
+  atkafka:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    container_name: atkafka
+    depends_on:
+      - kafka
+      - tap
+    ports:
+      # metrics port
+      - "6010:6009"
+    command: ["tap-mode"]
+    environment:
+      ATKAFKA_TAP_HOST: "ws://tap:2480"
+      ATKAFKA_DISABLE_ACKS: false
+      ATKAFKA_BOOTSTRAP_SERVERS: "kafka:29092"
+      ATKAFKA_OUTPUT_TOPIC: "tap-events"
+
+    restart: unless-stopped
+
+volumes:
+  zookeeper-data:
+  zookeeper-logs:
+  kafka-data:
+  tap-data:

go.mod

···
 go 1.25.4
 
 require (
+	github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de
 	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/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/sync v0.16.0
 	golang.org/x/time v0.6.0
 )
 
···
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	golang.org/x/crypto v0.40.0 // indirect
 	golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 // 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/xerrors v0.0.0-20231012003039-104605ab7028 // indirect

go.sum

···
 github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b/go.mod h1:4+EPqMRApwwE/6yo6CxiHoSnBzjRr3jsqer7frxP8y4=
 github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5 h1:iW0a5ljuFxkLGPNem5Ui+KBjFJzKg4Fv2fnxe4dvzpM=
 github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5/go.mod h1:Y2QMoi1vgtOIfc+6DhrMOGkLoGzqSV2rKp4Sm+opsyA=
+github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de h1:FxWPpzIjnTlhPwqqXc4/vE0f7GvRjuAsbW+HOIe8KnA=
+github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de/go.mod h1:DCaWoUhZrYW9p1lxo/cm8EmUOOzAPSEZNGF2DK1dJgw=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT8A=
 github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
···
 github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94=
 github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
 github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/mattn/go-runewidth v0.0.10/go.mod h1:RAqKPSqVFrSLVXbA8x7dzmKdmGzieGRCM46jaSJTDAk=
 github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o4kU=
 github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
 github.com/miekg/dns v1.1.50 h1:DQUfb9uc6smULcREF09Uc+/Gd46YWqJd5DbpPE9xkcA=
···
 github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
 github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
 github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
+github.com/rivo/uniseg v0.1.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
 github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/scylladb/termtables v0.0.0-20191203121021-c4c0b6d42ff4/go.mod h1:C1a7PQSMz9NShzorzCiG2fk9+xuCgLkPeCvMHYR2OWg=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
 github.com/smartystreets/assertions v1.2.0 h1:42S6lae5dvLc7BrLu/0ugRtcFVjoJNMC/N3yZFZkDFs=
 github.com/smartystreets/assertions v1.2.0/go.mod h1:tcbTF8ujkAEcZ8TElKY+i30BzYlVhC/LOxJk7iOWnoo=