appview: rework collaborators table, emit record

Signed-off-by: oppiliappan <[e-mail]>

appview/db/collaborators.go

···
+package db
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/bluesky-social/indigo/atproto/syntax"
+)
+
+type Collaborator struct {
+	// identifiers for the record
+	Id   int64
+	Did  syntax.DID
+	Rkey string
+
+	// content
+	SubjectDid syntax.DID
+	RepoAt     syntax.ATURI
+
+	// meta
+	Created time.Time
+}
+
+func AddCollaborator(e Execer, c Collaborator) error {
+	_, err := e.Exec(
+		`insert into collaborators (did, rkey, subject_did, repo_at) values (?, ?, ?, ?);`,
+		c.Did, c.Rkey, c.SubjectDid, c.RepoAt,
+	)
+	return err
+}
+
+func DeleteCollaborator(e Execer, filters ...filter) error {
+	var conditions []string
+	var args []any
+	for _, filter := range filters {
+		conditions = append(conditions, filter.Condition())
+		args = append(args, filter.Arg()...)
+	}
+
+	whereClause := ""
+	if conditions != nil {
+		whereClause = " where " + strings.Join(conditions, " and ")
+	}
+
+	query := fmt.Sprintf(`delete from collaborators %s`, whereClause)
+
+	_, err := e.Exec(query, args...)
+	return err
+}
+
+func CollaboratingIn(e Execer, collaborator string) ([]Repo, error) {
+	rows, err := e.Query(`select repo_at from collaborators where did = ?`, collaborator)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var repoAts []string
+	for rows.Next() {
+		var aturi string
+		err := rows.Scan(&aturi)
+		if err != nil {
+			return nil, err
+		}
+		repoAts = append(repoAts, aturi)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	if repoAts == nil {
+		return nil, nil
+	}
+
+	return GetRepos(e, 0, FilterIn("at_uri", repoAts))
+}

appview/db/db.go

···
 		return nil
 	})
 
+	// recreate and add rkey + created columns with default constraint
+	runMigration(db, "rework-collaborators-table", func(tx *sql.Tx) error {
+		// create new table
+		// - repo_at instead of repo integer
+		// - rkey field
+		// - created field
+		_, err := tx.Exec(`
+			create table collaborators_new (
+				-- identifiers for the record
+				id integer primary key autoincrement,
+				did text not null,
+				rkey text,
+
+				-- content
+				subject_did text not null,
+				repo_at text not null,
+
+				-- meta
+				created text default (strftime('%y-%m-%dt%h:%m:%sz', 'now')),
+
+				-- constraints
+				foreign key (repo_at) references repos(at_uri) on delete cascade
+			)
+		`)
+		if err != nil {
+			return err
+		}
+
+		// copy data
+		_, err = tx.Exec(`
+			insert into collaborators_new (id, did, rkey, subject_did, repo_at)
+			select
+				c.id,
+				r.did,
+				'',
+				c.did,
+				r.at_uri
+			from collaborators c
+			join repos r on c.repo = r.id
+		`)
+		if err != nil {
+			return err
+		}
+
+		// drop old table
+		_, err = tx.Exec(`drop table collaborators`)
+		if err != nil {
+			return err
+		}
+
+		// rename new table
+		_, err = tx.Exec(`alter table collaborators_new rename to collaborators`)
+		return err
+	})
+
 	return &DB{db}, nil
 }
 

appview/db/repos.go

···
 	return &repo, nil
 }
 
-func AddCollaborator(e Execer, collaborator, repoOwnerDid, repoName, repoKnot string) error {
-	_, err := e.Exec(
-		`insert into collaborators (did, repo)
-		values (?, (select id from repos where did = ? and name = ? and knot = ?));`,
-		collaborator, repoOwnerDid, repoName, repoKnot)
-	return err
-}
-
 func UpdateDescription(e Execer, repoAt, newDescription string) error {
 	_, err := e.Exec(
 		`update repos set description = ? where at_uri = ?`, newDescription, repoAt)
···
 	_, err := e.Exec(
 		`update repos set spindle = ? where at_uri = ?`, spindle, repoAt)
 	return err
-}
-
-func CollaboratingIn(e Execer, collaborator string) ([]Repo, error) {
-	rows, err := e.Query(`select repo from collaborators where did = ?`, collaborator)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-
-	var repoIds []int
-	for rows.Next() {
-		var id int
-		err := rows.Scan(&id)
-		if err != nil {
-			return nil, err
-		}
-		repoIds = append(repoIds, id)
-	}
-	if err := rows.Err(); err != nil {
-		return nil, err
-	}
-	if repoIds == nil {
-		return nil, nil
-	}
-
-	return GetRepos(e, 0, FilterIn("id", repoIds))
 }
 
 type RepoStats struct {

appview/repo/repo.go

···
 	"github.com/go-git/go-git/v5/plumbing"
 
 	comatproto "github.com/bluesky-social/indigo/api/atproto"
+	"github.com/bluesky-social/indigo/atproto/syntax"
 	lexutil "github.com/bluesky-social/indigo/lex/util"
 )
 
···
 		fail("You seem to be adding yourself as a collaborator.", nil)
 		return
 	}
-
 	l = l.With("collaborator", collaboratorIdent.Handle)
 	l = l.With("knot", f.Knot)
-	l.Info("adding to knot")
+
+	// announce this relation into the firehose, store into owners' pds
+	client, err := rp.oauth.AuthorizedClient(r)
+	if err != nil {
+		fail("Failed to write to PDS.", err)
+		return
+	}
+
+	// emit a record
+	currentUser := rp.oauth.GetUser(r)
+	rkey := tid.TID()
+	createdAt := time.Now()
+	resp, err := client.RepoPutRecord(r.Context(), &comatproto.RepoPutRecord_Input{
+		Collection: tangled.RepoCollaboratorNSID,
+		Repo:       currentUser.Did,
+		Rkey:       rkey,
+		Record: &lexutil.LexiconTypeDecoder{
+			Val: &tangled.RepoCollaborator{
+				Subject:   collaboratorIdent.DID.String(),
+				Repo:      string(f.RepoAt),
+				CreatedAt: createdAt.Format(time.RFC3339),
+			}},
+	})
+	// invalid record
+	if err != nil {
+		fail("Failed to write record to PDS.", err)
+		return
+	}
+	l = l.With("at-uri", resp.Uri)
+	l.Info("wrote record to PDS")
 
+	l.Info("adding to knot")
 	secret, err := db.GetRegistrationKey(rp.db, f.Knot)
 	if err != nil {
 		fail("Failed to add to knot.", err)
···
 		return
 	}
 
-	err = db.AddCollaborator(rp.db, collaboratorIdent.DID.String(), f.OwnerDid(), f.RepoName, f.Knot)
+	err = db.AddCollaborator(rp.db, db.Collaborator{
+		Did:        syntax.DID(currentUser.Did),
+		Rkey:       rkey,
+		SubjectDid: collaboratorIdent.DID,
+		RepoAt:     f.RepoAt,
+		Created:    createdAt,
+	})
 	if err != nil {
 		fail("Failed to add collaborator.", err)
 		return