appview: setup jetstream client

appview/db/db.go

···
 			primary key (user_did, subject_did),
 			check (user_did <> subject_did)
 		);
+		create table if not exists _jetstream (
+			id integer primary key autoincrement,
+			last_time_us integer not null
+		);
 	`)
 	if err != nil {
 		return nil, err

appview/db/follow.go

···
 	UserDid    string
 	SubjectDid string
 	FollowedAt *time.Time
-	AtUri      string
+	RKey       string
 }
 
-func (d *DB) AddFollow(userDid, subjectDid, atUri string) error {
-	query := `insert into follows (user_did, subject_did, at_uri) values (?, ?, ?)`
-	_, err := d.db.Exec(query, userDid, subjectDid, atUri)
+func (d *DB) AddFollow(userDid, subjectDid, rkey string) error {
+	query := `insert into follows (user_did, subject_did, rkey) values (?, ?, ?)`
+	_, err := d.db.Exec(query, userDid, subjectDid, rkey)
 	return err
 }
 
···
 
 	var follow Follow
 	var followedAt string
-	err := row.Scan(&follow.UserDid, &follow.SubjectDid, &followedAt, &follow.AtUri)
+	err := row.Scan(&follow.UserDid, &follow.SubjectDid, &followedAt, &follow.RKey)
 	if err != nil {
 		return nil, err
 	}

appview/db/jetstream.go

···
+package db
+
+func (d *DB) SaveLastTimeUs(lastTimeUs int64) error {
+	_, err := d.db.Exec(`insert into _jetstream (last_time_us) values (?)`, lastTimeUs)
+	return err
+}
+
+func (d *DB) GetLastTimeUs() (int64, error) {
+	var lastTimeUs int64
+	row := d.db.QueryRow(`select last_time_us from _jetstream`)
+	err := row.Scan(&lastTimeUs)
+	return lastTimeUs, err
+}

appview/state/state.go

···
 package state
 
 import (
+	"context"
 	"crypto/hmac"
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/json"
 	"fmt"
 	"log"
 	"net/http"
···
 	comatproto "github.com/bluesky-social/indigo/api/atproto"
 	"github.com/bluesky-social/indigo/atproto/syntax"
 	lexutil "github.com/bluesky-social/indigo/lex/util"
+	"github.com/bluesky-social/jetstream/pkg/models"
 	"github.com/go-chi/chi/v5"
 	tangled "github.com/sotangled/tangled/api/tangled"
 	"github.com/sotangled/tangled/appview"
 	"github.com/sotangled/tangled/appview/auth"
 	"github.com/sotangled/tangled/appview/db"
 	"github.com/sotangled/tangled/appview/pages"
+	"github.com/sotangled/tangled/jetstream"
 	"github.com/sotangled/tangled/rbac"
 )
 
···
 	tidClock *syntax.TIDClock
 	pages    *pages.Pages
 	resolver *appview.Resolver
+	jc       *jetstream.JetstreamClient
 }
 
 func Make() (*State, error) {
···
 
 	resolver := appview.NewResolver()
 
+	jc, err := jetstream.NewJetstreamClient("appview", []string{tangled.GraphFollowNSID}, nil, db)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create jetstream client: %w", err)
+	}
+	err = jc.StartJetstream(context.Background(), func(ctx context.Context, e *models.Event) error {
+		did := e.Did
+		raw := e.Commit.Record
+
+		switch e.Commit.Collection {
+		case tangled.GraphFollowNSID:
+			record := tangled.GraphFollow{}
+			err := json.Unmarshal(raw, &record)
+			if err != nil {
+				return err
+			}
+			err = db.AddFollow(did, record.Subject, e.Commit.RKey)
+			if err != nil {
+				return fmt.Errorf("failed to add follow to db: %w", err)
+			}
+		}
+
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to start jetstream watcher: %w", err)
+	}
+
 	state := &State{
 		db,
-		auth, enforcer, clock, pgs, resolver,
+		auth,
+		enforcer,
+		clock,
+		pgs,
+		resolver,
+		jc,
 	}
 
 	return state, nil
···
 	switch r.Method {
 	case http.MethodPost:
 		createdAt := time.Now().Format(time.RFC3339)
+		rkey := s.TID()
 		resp, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{
 			Collection: tangled.GraphFollowNSID,
 			Repo:       currentUser.Did,
-			Rkey:       s.TID(),
+			Rkey:       rkey,
 			Record: &lexutil.LexiconTypeDecoder{
 				Val: &tangled.GraphFollow{
 					Subject:   subjectIdent.DID.String(),
···
 			return
 		}
 
-		err = s.db.AddFollow(currentUser.Did, subjectIdent.DID.String(), resp.Uri)
+		err = s.db.AddFollow(currentUser.Did, subjectIdent.DID.String(), rkey)
 		if err != nil {
 			log.Println("failed to follow", err)
 			return
···
 			return
 		}
 
-		existingRecordUri, _ := syntax.ParseATURI(follow.AtUri)
-
 		resp, err := comatproto.RepoDeleteRecord(r.Context(), client, &comatproto.RepoDeleteRecord_Input{
 			Collection: tangled.GraphFollowNSID,
 			Repo:       currentUser.Did,
-			Rkey:       existingRecordUri.RecordKey().String(),
+			Rkey:       follow.RKey,
 		})
 
 		log.Println(resp.Commit.Cid)