atproto: save lexicon commit events

pkg/atproto/lexicon_repo.go

···
 	"io"
 	"io/fs"
 	"strings"
+	"time"
 
+	comatproto "github.com/bluesky-social/indigo/api/atproto"
 	atcrypto "github.com/bluesky-social/indigo/atproto/crypto"
 	"github.com/bluesky-social/indigo/atproto/data"
 	"github.com/bluesky-social/indigo/atproto/lexicon"
···
 	"github.com/bluesky-social/indigo/models"
 	"github.com/bluesky-social/indigo/mst"
 	atrepo "github.com/bluesky-social/indigo/repo"
+	"github.com/bluesky-social/indigo/util"
 	"github.com/ipfs/go-cid"
 	cbg "github.com/whyrusleeping/cbor-gen"
 	"gorm.io/driver/sqlite"
···
 	"stream.place/streamplace/lexicons"
 	"stream.place/streamplace/pkg/config"
 	"stream.place/streamplace/pkg/log"
+	"stream.place/streamplace/pkg/model"
 )
 
 var LexiconRepo *atrepo.Repo
 var LexiconPubMultibase string
 var RepoUser models.Uid = models.Uid(1)
 var CarStore carstore.CarStore
+var ActionCreate = "create"
+var ActionUpdate = "update"
+var ActionDelete = "delete"
 
 func walkLexicons(ctx context.Context, bundle fs.FS, path string) ([][]byte, error) {
 	ret := [][]byte{}
···
 	Close() error
 }
 
-func MakeLexiconRepo(ctx context.Context, cli *config.CLI) (Closer, error) {
+func MakeLexiconRepo(ctx context.Context, cli *config.CLI, mod model.Model) (Closer, error) {
 	ctx = log.WithLogValues(ctx, "func", "MakeLexiconRepo")
 	fd, err := cli.DataFileCreate([]string{"carstore", "empty"}, true)
 	if err != nil {
···
 	if err != nil {
 		return nil, fmt.Errorf("failed to walk lexicon files: %w", err)
 	}
+
+	ops := []*comatproto.SyncSubscribeRepos_RepoOp{}
+
 	for _, lex := range lexs {
 		lexFile := lexicon.SchemaFile{}
 		err := json.Unmarshal(lex, &lexFile)
···
 		if err != nil {
 			return nil, err
 		}
+		cidLink := lexutil.LexLink(*newCid)
 
 		oldCid, _, err := LexiconRepo.GetRecord(ctx, rpath)
 		if errors.Is(err, mst.ErrNotFound) {
···
 				return nil, err
 			}
 			log.Log(ctx, "created new lexicon record", "rpath", rpath, "cid", newCid.String())
+			ops = append(ops, &comatproto.SyncSubscribeRepos_RepoOp{
+				Action: ActionCreate,
+				Path:   rpath,
+				Cid:    &cidLink,
+			})
 		} else if err != nil {
 			return nil, err
 		} else {
···
 				if err != nil {
 					return nil, err
 				}
+				oldLink := lexutil.LexLink(oldCid)
+				ops = append(ops, &comatproto.SyncSubscribeRepos_RepoOp{
+					Action: ActionUpdate,
+					Path:   rpath,
+					Prev:   &oldLink,
+					Cid:    &cidLink,
+				})
 			}
 		}
 		currentRoot, currentRev, err = LexiconRepo.Commit(ctx, signer)
 		if err != nil {
 			return nil, fmt.Errorf("failed to commit: %w", err)
 		}
+
 		log.Log(ctx, "LexiconRepo committed", "cid", currentRoot.String(), "rev", currentRev)
 	}
-	_, err = ses.CloseWithRoot(ctx, currentRoot, currentRev)
+	blocks, err := ses.CloseWithRoot(ctx, currentRoot, currentRev)
 	if err != nil {
 		return nil, fmt.Errorf("failed to close delta session: %w", err)
+	}
+	if len(ops) > 0 {
+		commit := &comatproto.SyncSubscribeRepos_Commit{
+			Repo:   cli.MyDID(),
+			Blocks: blocks,
+			Rev:    currentRev,
+			// Since:  currentRev,
+			Commit: lexutil.LexLink(currentRoot),
+			Time:   time.Now().Format(util.ISO8601),
+			Ops:    ops,
+			TooBig: false,
+		}
+		err := mod.CreateCommitEvent(commit)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create commit event: %w", err)
+		}
 	}
 
 	return sqlDB, nil

pkg/atproto/lexicon_repo_test.go

···
 	"io/fs"
 	"testing"
 	"testing/fstest"
+	"time"
 
 	"github.com/stretchr/testify/require"
 	"stream.place/streamplace/lexicons"
 	"stream.place/streamplace/pkg/config"
+	"stream.place/streamplace/pkg/model"
 )
 
 func TestLexiconRepo(t *testing.T) {
-	cli := config.CLI{}
+	cli := config.CLI{
+		PublicHost: "example.com",
+	}
 	cli.DataDir = t.TempDir()
+	mod, err := model.MakeDB(":memory:")
+	require.NoError(t, err)
 
 	// creating a new repo
-	handle, err := MakeLexiconRepo(context.Background(), &cli)
+	handle, err := MakeLexiconRepo(context.Background(), &cli, mod)
 	require.NoError(t, err)
 	r, sess, err := OpenLexiconRepo(context.Background())
 	require.NoError(t, err)
···
 	require.NotNil(t, rec)
 	handle.Close()
 
+	evts, err := mod.GetCommitEventsSince(cli.MyDID(), time.Time{})
+	require.NoError(t, err)
+	require.Len(t, evts, 1)
+	require.Equal(t, evts[0].RepoDID, cli.MyDID())
+
 	// opening an existing repo
-	handle, err = MakeLexiconRepo(context.Background(), &cli)
+	handle, err = MakeLexiconRepo(context.Background(), &cli, mod)
 	require.NoError(t, err)
 	handle.Close()
 
···
 	AllFiles = modifiedFS
 
 	// opening an existing repo with modified lexicon
-	handle, err = MakeLexiconRepo(context.Background(), &cli)
+	handle, err = MakeLexiconRepo(context.Background(), &cli, mod)
 	require.NoError(t, err)
 	handle.Close()
 
-	// Now modifiedFS is a fs.FS with the first file modified
+	evts, err = mod.GetCommitEventsSince(cli.MyDID(), time.Time{})
+	require.NoError(t, err)
+	require.Len(t, evts, 2)
+	require.Equal(t, evts[0].RepoDID, cli.MyDID())
+	require.Equal(t, evts[1].RepoDID, cli.MyDID())
+	commit, err := evts[1].ToCommitEvent()
+	require.NoError(t, err)
+	require.Equal(t, commit.Since, &evts[0].CID)
 }

pkg/cmd/streamplace.go

···
 	if err != nil {
 		return fmt.Errorf("error creating streamplace dir at %s:%w", cli.DataDir, err)
 	}
-	handle, err := atproto.MakeLexiconRepo(ctx, &cli)
-	if err != nil {
-		return err
-	}
-	defer handle.Close()
 	schema, err := v0.MakeV0Schema()
 	if err != nil {
 		return err
···
 	if err != nil {
 		return err
 	}
+	handle, err := atproto.MakeLexiconRepo(ctx, &cli, mod)
+	if err != nil {
+		return err
+	}
+	defer handle.Close()
 	var noter notifications.FirebaseNotifier
 	if cli.FirebaseServiceAccount != "" {
 		noter, err = notifications.MakeFirebaseNotifier(ctx, cli.FirebaseServiceAccount)

pkg/model/model.go

···
 	"strings"
 	"time"
 
+	comatproto "github.com/bluesky-social/indigo/api/atproto"
 	"github.com/bluesky-social/indigo/api/bsky"
 	"github.com/lmittmann/tint"
 	slogGorm "github.com/orandin/slog-gorm"
···
 	UpdateServerSettings(ctx context.Context, settings *ServerSettings) error
 	GetServerSettings(ctx context.Context, server string, repoDID string) (*ServerSettings, error)
 	DeleteServerSettings(ctx context.Context, server string, repoDID string) error
+
+	CreateCommitEvent(commit *comatproto.SyncSubscribeRepos_Commit) error
+	GetCommitEventsSince(repoDID string, t time.Time) ([]*XrpcStreamEvent, error)
+	GetMostRecentCommitEvent(repoDID string) (*XrpcStreamEvent, error)
 }
 
 func MakeDB(dbURL string) (Model, error) {
···
 		ChatProfile{},
 		oatproxy.OAuthSession{},
 		ServerSettings{},
+		XrpcStreamEvent{},
 	} {
 		err = db.AutoMigrate(model)
 		if err != nil {

pkg/model/xrpc_stream_event.go

···
+package model
+
+import (
+	"bytes"
+	"errors"
+	"time"
+
+	comatproto "github.com/bluesky-social/indigo/api/atproto"
+	"github.com/bluesky-social/indigo/util"
+	"gorm.io/gorm"
+)
+
+type XrpcStreamEvent struct {
+	CID       string    `json:"cid" gorm:"primaryKey"`
+	RepoDID   string    `json:"repoDID" gorm:"index:idx_repo_timestamp,priority:1;column:repo_did"`
+	Timestamp time.Time `json:"timestamp" gorm:"index:idx_repo_timestamp,priority:2;column:timestamp"`
+	Data      []byte    `json:"data"`
+}
+
+func (ev *XrpcStreamEvent) ToCommitEvent() (*comatproto.SyncSubscribeRepos_Commit, error) {
+	commit := &comatproto.SyncSubscribeRepos_Commit{}
+	err := commit.UnmarshalCBOR(bytes.NewReader(ev.Data))
+	if err != nil {
+		return nil, err
+	}
+	return commit, nil
+}
+
+func (m *DBModel) CreateCommitEvent(commit *comatproto.SyncSubscribeRepos_Commit) error {
+	prev, err := m.GetMostRecentCommitEvent(commit.Repo)
+	if err != nil {
+		return err
+	}
+	if prev != nil {
+		commit.Since = &prev.CID
+	}
+	buf := bytes.Buffer{}
+	err = commit.MarshalCBOR(&buf)
+	if err != nil {
+		return err
+	}
+	timestamp, err := time.Parse(util.ISO8601, commit.Time)
+	if err != nil {
+		return err
+	}
+	event := &XrpcStreamEvent{
+		CID:       commit.Commit.String(),
+		RepoDID:   commit.Repo,
+		Timestamp: timestamp.UTC(),
+		Data:      buf.Bytes(),
+	}
+	return m.DB.Create(event).Error
+}
+
+func (m *DBModel) GetCommitEventsSince(repoDID string, t time.Time) ([]*XrpcStreamEvent, error) {
+	var events []*XrpcStreamEvent
+	query := m.DB.Where("repo_did = ?", repoDID)
+	query = query.Where("timestamp > ?", t.UTC())
+	err := query.Order("timestamp ASC").Find(&events).Error
+	if err != nil {
+		return nil, err
+	}
+	return events, nil
+}
+
+func (m *DBModel) GetMostRecentCommitEvent(repoDID string) (*XrpcStreamEvent, error) {
+	var event XrpcStreamEvent
+	err := m.DB.Where("repo_did = ?", repoDID).
+		Order("timestamp DESC").
+		Limit(1).
+		First(&event).Error
+	if errors.Is(err, gorm.ErrRecordNotFound) {
+		return nil, nil
+	} else if err != nil {
+		return nil, err
+	}
+	return &event, nil
+}