diff --git a/appview/db/db.go b/appview/db/db.go --- a/appview/db/db.go +++ b/appview/db/db.go @@ -1262,6 +1262,108 @@ `) return err }) + runMigration(conn, logger, "rewrite-pulls-table", func(tx *sql.Tx) error { + _, err := tx.Exec(` + create table pulls2 ( + -- identifiers + id integer primary key autoincrement, + pull_id integer not null, + did text not null, + rkey text not null, + at_uri text generated always as ('at://' || did || '/' || 'sh.tangled.repo.pull' || '/' || rkey) stored, + + -- content + state integer not null default 0 check (state in (0, 1, 2, 3)), -- closed, open, merged, deleted + + unique(did, rkey) + ); + + create table pull_rounds ( + -- identifiers + id integer primary key autoincrement, + did text not null, + rkey text not null, + cid text not null, + + target_repo_at text not null, + target_branch text not null, + source_repo_at text, + source_branch text, + patch text not null, + title text not null, + body text not null, + + created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + + unique(target_repo_at, pull_id, cid), + unique(did, rkey, cid), + + foreign key (did, rkey) + referneces pulls2(did, rkey) + on delete cascade + ); + + create view pull_rounds_view as + select + pr.id, + pr.cid, + + p.pull_id, + p.did, + p.rkey, + p.at_uri, + p.state, + + row_number() over ( + partition by p.did, p.rkey + order by pr.id + ) -1 as round, + + pr.target_repo_at, + pr.target_branch, + pr.source_repo_at, + pr.source_branch, + pr.patch, + pr.title, + pr.body, + pr.created + from pull_rounds pr + join pulls p + on p.did = pr.did + and p.rkey = pr.rkey; + + create view pull_latests_view as + select * + from ( + select + *, + (count(*) over (partition by did, rkey)) - 1 as round, + row_number() over ( + partition by did, rkey + order by id desc + ) as rn + from pull_rounds_view + ) + where rn = 1; + + create table reference_links_new ( + id integer primary key autoincrement, + from_at text not null, + from_cid text, + to_at text not null, + unique (from_at, from_cid, to_at) + ); + + insert into reference_links_new + select * from reference_links; + + drop table reference_links; + + alter table reference_links_new to reference_links; + `) + return err + }) + return &DB{ db, logger, diff --git a/appview/db/pulls2.go b/appview/db/pulls2.go new file mode 100644 --- /dev/null +++ b/appview/db/pulls2.go @@ -0,0 +1,235 @@ +package db + +import ( + "database/sql" + "fmt" + "strings" + "time" + + "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/appview/models" + "tangled.org/core/appview/pagination" +) + +// NewPullRound creates new PR submission. +// Set pullId to open a new PR. +func NewPullRound(tx *sql.Tx, pullId int, round *models.PullRound) error { + // Create new PR when pullId isn't set + if pullId == 0 { + // ensure sequence exists + _, err := tx.Exec(` + insert or ignore into repo_pull_seqs (repo_at, next_pull_id) + values (?, 1) + `, round.Target.RepoAt) + if err != nil { + return err + } + + err = tx.QueryRow(` + update repo_pull_seqs + set next_pull_id = next_pull_id + 1 + where repo_at = ? + returning next_pull_id - 1 + `, round.Target.RepoAt).Scan(&pullId) + if err != nil { + return err + } + + _, err = tx.Exec( + `insert into pulls2 (pull_id, did, rkey, state) + values (?, ?, ?, ?)`, + pullId, + round.Did, + round.Rkey, + models.PullOpen, + ) + if err != nil { + return fmt.Errorf("insert pull submission: %w", err) + } + } + + var sourceRepoAt, sourceBranch *string + if round.Source != nil { + x := round.Source.RepoAt.String() + sourceRepoAt = &x + sourceBranch = &round.Source.Branch + } + _, err := tx.Exec( + `insert into pull_rounds ( + pull_did, + pull_rkey, + cid, + target_repo_at, + target_branch, + source_repo_at, + source_branch, + patch, + title, + body, + created + ) + values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + round.Did, + round.Rkey, + round.Cid, + round.Target.RepoAt, + round.Target.Branch, + sourceRepoAt, + sourceBranch, + round.Patch, + round.Title, + round.Body, + round.Created.Format(time.RFC3339), + ) + if err != nil { + return fmt.Errorf("insert pull submission: %w", err) + } + + err = tx.QueryRow( + `select id, round from pull_rounds_view + where did = ? and rkey = ? and cid = ?`, + round.Did, round.Rkey, round.Cid, + ).Scan(&round.Id, &round.Round) + if err != nil { + return fmt.Errorf("get id and round number: %w", err) + } + + if err := putReferences(tx, round.AtUri(), &round.Cid, round.References); err != nil { + return fmt.Errorf("put reference_links: %w", err) + } + + return nil +} + +func SetPullState2(e Execer, pullAt syntax.ATURI, state models.PullState) error { + _, err := e.Exec( + `update pulls2 set state = ? where at_uri = ? and (state <> ? or state <> ?)`, + state, + pullAt, + models.PullDeleted, // only update state of non-deleted pulls + models.PullMerged, // only update state of non-merged pulls + ) + return err +} + +func ClosePull2(e Execer, pullAt syntax.ATURI) error { + return SetPullState2(e, pullAt, models.PullClosed) +} + +func ReopenPull2(e Execer, pullAt syntax.ATURI) error { + return SetPullState2(e, pullAt, models.PullOpen) +} + +func MergePull2(e Execer, pullAt syntax.ATURI) error { + return SetPullState2(e, pullAt, models.PullMerged) +} + +func DeletePull2(e Execer, pullAt syntax.ATURI) error { + return SetPullState2(e, pullAt, models.PullDeleted) +} + +func GetPulls2(e Execer, filters ...filter) ([]*models.Pull2, error) { + return GetPullsPaginated(e, pagination.Page{}, filters...) +} + +func GetPullsPaginated(e Execer, page pagination.Page, filters ...filter) ([]*models.Pull2, error) { + pullsMap := make(map[syntax.ATURI]*models.Pull2) + + 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 ") + } + + pLower := FilterGte("row_num", page.Offset+1) + pUpper := FilterLte("row_num", page.Offset+page.Limit) + + pageClause := "" + if page.Limit > 0 { + args = append(args, pLower.Arg()...) + args = append(args, pUpper.Arg()...) + pageClause = " where " + pLower.Condition() + " and " + pUpper.Condition() + } + + query := fmt.Sprintf( + `select * from ( + select + id, + did, + rkey, + cid, + target_repo_at, + target_branch, + source_repo_at, + source_branch, + patch, + title, + body, + created, + row_number() over (order by id desc) as row_num + from + pull_rounds + %s + ) ranked_pull_rounds + %s`, + whereClause, + pageClause, + ) + + rows, err := e.Query(query, args...) + if err != nil { + return nil, fmt.Errorf("query pulls table: %w", err) + } + defer rows.Close() + + for rows.Next() { + var round models.PullRound + var sourceBranch, sourceRepoAt sql.NullString + var created string + err := rows.Scan( + &round.Id, + &round.Did, + &round.Rkey, + &round.Cid, + &round.Target.RepoAt, + &round.Target.Branch, + &sourceBranch, + &sourceRepoAt, + &round.Patch, + &round.Title, + &round.Body, + &created, + ) + if err != nil { + return nil, fmt.Errorf("scan row: %w", err) + } + + if sourceBranch.Valid && sourceRepoAt.Valid { + round.Source = &models.PullSource2{} + round.Source.Branch = sourceBranch.String + round.Source.RepoAt = syntax.ATURI(sourceRepoAt.String) + } + createdAtTime, _ := time.Parse(time.RFC3339, created) + round.Created = createdAtTime + + pull, ok := pullsMap[round.AtUri()] + if !ok { + pull = &models.Pull2{ + Did: round.Did, + Rkey: round.Rkey, + } + } + pull.Submissions = append(pull.Submissions, &round) + pullsMap[round.AtUri()] = pull + } + + // TODO: fetch pulls (id, pull_id, state) from (did, rkey) + + panic("unimplemented") +} diff --git a/appview/db/reference.go b/appview/db/reference.go --- a/appview/db/reference.go +++ b/appview/db/reference.go @@ -161,23 +161,26 @@ return uris, nil } func putReferences(tx *sql.Tx, fromAt syntax.ATURI, fromCid *syntax.CID, references []syntax.ATURI) error { - err := deleteReferences(tx, fromAt) - if err != nil { - return fmt.Errorf("delete old reference_links: %w", err) - } - if len(references) == 0 { - return nil + // clear existing referneces when cid isn't provided + if fromCid == nil { + err := deleteReferences(tx, fromAt) + if err != nil { + return fmt.Errorf("delete old reference_links: %w", err) + } + if len(references) == 0 { + return nil + } } values := make([]string, 0, len(references)) args := make([]any, 0, len(references)*2) for _, ref := range references { - values = append(values, "(?, ?)") - args = append(args, fromAt, ref) + values = append(values, "(?, ?, ?)") + args = append(args, fromAt, fromCid, ref) } - _, err = tx.Exec( + _, err := tx.Exec( fmt.Sprintf( - `insert into reference_links (from_at, to_at) + `insert into reference_links (from_at, from_cid, to_at) values %s`, strings.Join(values, ","), ), diff --git a/appview/models/pull2.go b/appview/models/pull2.go new file mode 100644 --- /dev/null +++ b/appview/models/pull2.go @@ -0,0 +1,188 @@ +package models + +import ( + "fmt" + "time" + + "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/api/tangled" +) + +type Pull2 struct { + // TODO: PullId is not atprotated yet + PullId int + Did syntax.DID + Rkey syntax.RecordKey + + Submissions []*PullRound // PR submission history + + // backlinked records + + State PullState + Labels LabelState + Comments []Comment + + // optionally, populate these when querying for reverse mappings + + Repo *Repo +} + +// PullRound represents snapshot of `sh.tangled.repo.pull` record +// NOTE: Non-patch change can make new submission +type PullRound struct { + Id int64 + Round int + Did syntax.DID + Rkey syntax.RecordKey + Cid syntax.CID + + // content + + Target PullTarget + Source *PullSource2 // optional for patch based PR + Patch string // list of commits encoded in patch object + Title string + Body string + Mentions []syntax.DID + References []syntax.ATURI + Created time.Time +} + +type PullTarget struct { + RepoAt syntax.ATURI + Branch string +} + +// PullSource2 is not a source of truth but rather a metadata to help resubmitting +type PullSource2 struct { + RepoAt syntax.ATURI + Branch string +} + +func (p Pull2) AsRecord() tangled.RepoPull { + return p.Latest().AsRecord() +} + +func (p *Pull2) AtUri() syntax.ATURI { + return syntax.ATURI(fmt.Sprintf("at://%s/%s/%s", p.Did, tangled.RepoPullNSID, p.Rkey)) +} + +func (p *Pull2) Latest() *PullRound { + return p.Submissions[len(p.Submissions)-1] +} + +func (p *Pull2) CID() syntax.CID { + return p.Latest().Cid +} + +func (p *Pull2) Created() time.Time { + return p.Submissions[0].Created +} + +func (p *Pull2) Edited() time.Time { + return p.Latest().Created +} + +func (p *Pull2) RepoAt() syntax.ATURI { + return p.Latest().Target.RepoAt +} + +func (p *Pull2) Validate() error { + if len(p.Submissions) == 0 { + return fmt.Errorf("Pull should include at least one submission") + } + // we don't need to validate existing records + if err := p.Latest().Validate(); err != nil { + return fmt.Errorf("validate latest stack: %w", err) + } + return nil +} + +func PullContentFromRecord(cid syntax.CID, record tangled.RepoPull) PullRound { + var source *PullSource2 + if record.Source != nil { + source = &PullSource2{} + source.Branch = record.Source.Branch + if record.Source.Repo != nil { + source.RepoAt = syntax.ATURI(*record.Source.Repo) + } else { + source.RepoAt = syntax.ATURI(record.Target.Repo) + } + } + var ( + body string + mentions = make([]syntax.DID, len(record.Mentions)) + references = make([]syntax.ATURI, len(record.References)) + ) + if record.Body != nil { + body = *record.Body + } + for i, v := range record.Mentions { + mentions[i] = syntax.DID(v) + } + for i, v := range record.References { + references[i] = syntax.ATURI(v) + } + created, err := time.Parse(record.CreatedAt, time.RFC3339) + if err != nil { + created = time.Now() + } + return PullRound{ + Cid: cid, + Target: PullTarget{ + RepoAt: syntax.ATURI(record.Target.Repo), + Branch: record.Target.Branch, + }, + Source: source, + Patch: record.Patch, + Title: record.Title, + Body: body, + Created: created, + } +} + +// NOTE: AsRecord doesn't include the patch blob id in returned atproto record +func (p PullRound) AsRecord() tangled.RepoPull { + var source *tangled.RepoPull_Source + + if p.Source != nil { + repoAt := p.Source.RepoAt.String() + source = &tangled.RepoPull_Source{ + Repo: &repoAt, + Branch: p.Source.Branch, + } + } + + var body *string + if p.Body != "" { + body = &p.Body + } + return tangled.RepoPull{ + Target: &tangled.RepoPull_Target{ + Repo: p.Target.RepoAt.String(), + Branch: p.Target.Branch, + }, + Source: source, + Title: p.Title, + Body: body, + Mentions: toStringList(p.Mentions), + References: toStringList(p.References), + } +} + +func (p *PullRound) AtUri() syntax.ATURI { + return syntax.ATURI(fmt.Sprintf("at://%s/%s/%s", p.Did, tangled.RepoPullNSID, p.Rkey)) +} + +func (p *PullRound) Validate() error { + // TODO: validate patches + return nil +} + +func toStringList[T fmt.Stringer](list []T) []string { + slist := make([]string, len(list)) + for i, v := range list { + slist[i] = v.String() + } + return slist +} diff --git a/appview/service/pull/errors.go b/appview/service/pull/errors.go new file mode 100644 --- /dev/null +++ b/appview/service/pull/errors.go @@ -0,0 +1,12 @@ +package pull + +import "errors" + +var ( + ErrUnAuthenticated = errors.New("user session missing") + ErrForbidden = errors.New("unauthorized operation") + ErrDatabaseFail = errors.New("db op fail") + ErrPDSFail = errors.New("pds op fail") + ErrIndexerFail = errors.New("indexer fail") + ErrValidationFail = errors.New("pull validation fail") +) diff --git a/appview/service/pull/merge.go b/appview/service/pull/merge.go new file mode 100644 --- /dev/null +++ b/appview/service/pull/merge.go @@ -0,0 +1,41 @@ +package pull + +import ( + "context" + + "tangled.org/core/appview/models" + "tangled.org/core/appview/pages/repoinfo" + "tangled.org/core/appview/session" +) + +func (s *Service) MergePull(ctx context.Context, pull *models.Pull2) error { + l := s.logger.With("method", "MergePull") + sess := session.FromContext(ctx) + if sess == nil { + l.Error("user session is missing in context") + return ErrForbidden + } + sessDid := sess.Data.AccountDID + l = l.With("did", sessDid) + + // TODO: make this more granular + roles := repoinfo.RolesInRepo{Roles: s.enforcer.GetPermissionsInRepo(sessDid.String(), pull.Repo.Knot, pull.Repo.DidSlashRepo())} + isRepoOwner := roles.IsOwner() + isCollaborator := roles.IsCollaborator() + isAuthor := sessDid == pull.Did + if !(isRepoOwner || isCollaborator || isAuthor) { + l.Error("user is not authorized") + return ErrForbidden + } + + // 1. request knot to apply series of patches + // 2. create new pull.state record + // 3. update db + + panic("unimplemented") + + pull.State = models.PullMerged + + // s.notifier.NewPullState(ctx, sessDid, pull) + return nil +} diff --git a/appview/service/pull/pull.go b/appview/service/pull/pull.go new file mode 100644 --- /dev/null +++ b/appview/service/pull/pull.go @@ -0,0 +1,239 @@ +package pull + +import ( + "context" + "log/slog" + "time" + + "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/syntax" + lexutil "github.com/bluesky-social/indigo/lex/util" + "tangled.org/core/api/tangled" + "tangled.org/core/appview/config" + "tangled.org/core/appview/db" + pulls_indexer "tangled.org/core/appview/indexer/pulls" + "tangled.org/core/appview/mentions" + "tangled.org/core/appview/models" + "tangled.org/core/appview/notify" + "tangled.org/core/appview/session" + "tangled.org/core/idresolver" + "tangled.org/core/rbac" + "tangled.org/core/tid" +) + +type Service struct { + config *config.Config + db *db.DB + enforcer *rbac.Enforcer + indexer *pulls_indexer.Indexer + logger *slog.Logger + notifier notify.Notifier + idResolver *idresolver.Resolver + mentionsResolver *mentions.Resolver +} + +func NewService( + logger *slog.Logger, + config *config.Config, + db *db.DB, + enforcer *rbac.Enforcer, + notifier notify.Notifier, + idResolver *idresolver.Resolver, + mentionsResolver *mentions.Resolver, + indexer *pulls_indexer.Indexer, +) Service { + return Service{ + config, + db, + enforcer, + indexer, + logger, + notifier, + idResolver, + mentionsResolver, + } +} + +// SubmitPull creates a new PR or resubmits existing PR. +// `pull` can be `nil` for creating a new PR. +func (s *Service) SubmitPull( + ctx context.Context, + pull *models.Pull2, + target models.PullTarget, + source *models.PullSource2, + patch, title, body string, +) error { + l := s.logger.With("method", "NewPullSubmission") + sess := session.FromContext(ctx) + if sess == nil { + l.Error("user session is missing in context") + return ErrForbidden + } + sessDid := sess.Data.AccountDID + l = l.With("did", sessDid) + + var ( + did syntax.DID + rkey syntax.RecordKey + ) + if pull == nil { + // new pr + did = sessDid + rkey = syntax.RecordKey(tid.TID()) + } else { + // resubmit + if sessDid != pull.Did { + l.Error("only author can edit the pull") + return ErrForbidden + } + did = pull.Did + rkey = pull.Rkey + } + + mentions, references := s.mentionsResolver.Resolve(ctx, body) + + round := models.PullRound{ + Did: did, + Rkey: rkey, + Target: target, + Source: source, + Patch: patch, + Title: title, + Body: body, + Mentions: mentions, + References: references, + Created: time.Now(), + } + if err := round.Validate(); err != nil { + l.Error("validation error", "err", err) + return ErrValidationFail + } + + atpclient := sess.APIClient() + record := round.AsRecord() + + var exCid *string + if pull != nil { + x := pull.CID().String() + exCid = &x + } + resp, err := atproto.RepoPutRecord(ctx, atpclient, &atproto.RepoPutRecord_Input{ + Collection: tangled.RepoPullNSID, + SwapRecord: exCid, + Record: &lexutil.LexiconTypeDecoder{ + Val: &record, + }, + }) + if err != nil { + l.Error("atproto.RepoPutRecord failed", "err", err) + return ErrPDSFail + } + round.Cid = syntax.CID(resp.Cid) + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + l.Error("db.BeginTx failed", "err", err) + return ErrDatabaseFail + } + defer tx.Rollback() + + if err := db.NewPullRound(tx, 0, &round); err != nil { + l.Error("db.UpdatePull2 failed", "err", err) + return ErrDatabaseFail + } + + if err = tx.Commit(); err != nil { + l.Error("tx.Commit failed", "err", err) + return ErrDatabaseFail + } + + if pull == nil { + // s.notifier.NewPull(ctx, &round) + } else { + pull.Submissions = append(pull.Submissions, &round) + // s.notifier.ResubmitPull(ctx, &round) + } + + return nil +} + +func (s *Service) DeletePull(ctx context.Context, pull *models.Pull2) error { + l := s.logger.With("method", "DeletePull") + sess := session.FromContext(ctx) + if sess == nil { + l.Error("user session is missing in context") + return ErrForbidden + } + sessDid := sess.Data.AccountDID + l = l.With("did", sessDid) + + if sessDid != pull.Did { + l.Error("only author can delete the pull") + return ErrForbidden + } + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + l.Error("db.BeginTx failed", "err", err) + return ErrDatabaseFail + } + defer tx.Rollback() + + if err := db.DeletePull2(tx, pull.AtUri()); err != nil { + l.Error("db.DeletePull2 failed", "err", err) + return ErrDatabaseFail + } + + atpclient := sess.APIClient() + _, err = atproto.RepoDeleteRecord(ctx, atpclient, &atproto.RepoDeleteRecord_Input{ + Collection: tangled.RepoIssueNSID, + Repo: pull.Did.String(), + Rkey: pull.Rkey.String(), + }) + if err != nil { + l.Error("atproto.RepoDeleteRecord failed", "err", err) + return ErrPDSFail + } + + if err := tx.Commit(); err != nil { + l.Error("tx.Commit failed", "err", err) + return ErrDatabaseFail + } + + pull.State = models.PullDeleted + + // s.notifier.DeletePull(ctx, pull) + return nil +} + +func (s *Service) ListPulls(ctx context.Context, repo *models.Repo, searchOpts models.PullSearchOptions) ([]*models.Pull2, error) { + l := s.logger.With("method", "ListPulls") + + var pulls []*models.Pull2 + var err error + if searchOpts.Keyword != "" { + res, err := s.indexer.Search(ctx, searchOpts) + if err != nil { + l.Error("failed to search for pulls", "err", err) + return nil, ErrIndexerFail + } + l.Debug("searched pulls with indexer", "count", len(res.Hits)) + pulls, err = db.GetPulls2(s.db, db.FilterIn("id", res.Hits)) + if err != nil { + l.Error("failed to get pulls", "err", err) + return nil, ErrDatabaseFail + } + } else { + pulls, err = db.GetPullsPaginated( + s.db, + searchOpts.Page, + db.FilterEq("repo_at", repo.RepoAt()), + db.FilterEq("state", searchOpts.State), + ) + if err != nil { + l.Error("failed to get pulls", "err", err) + return nil, ErrDatabaseFail + } + } + return pulls, nil +} diff --git a/appview/service/pull/state.go b/appview/service/pull/state.go new file mode 100644 --- /dev/null +++ b/appview/service/pull/state.go @@ -0,0 +1,47 @@ +package pull + +import ( + "context" + + "tangled.org/core/appview/db" + "tangled.org/core/appview/models" + "tangled.org/core/appview/pages/repoinfo" + "tangled.org/core/appview/session" +) + +func (s *Service) ClosePull(ctx context.Context, pull *models.Pull2) error { + l := s.logger.With("method", "CloseIssue") + sess := session.FromContext(ctx) + if sess == nil { + l.Error("user session is missing in context") + return ErrUnAuthenticated + } + sessDid := sess.Data.AccountDID + l = l.With("did", sessDid) + + // TODO: make this more granular + roles := repoinfo.RolesInRepo{Roles: s.enforcer.GetPermissionsInRepo(sessDid.String(), pull.Repo.Knot, pull.Repo.DidSlashRepo())} + isRepoOwner := roles.IsOwner() + isCollaborator := roles.IsCollaborator() + isAuthor := sessDid == pull.Did + if !(isRepoOwner || isCollaborator || isAuthor) { + l.Error("user is not authorized") + return ErrForbidden + } + + err := db.ClosePull2(s.db, pull.AtUri()) + if err != nil { + l.Error("db.ClosePull2 failed", "err", err) + return ErrDatabaseFail + } + + // change the issue state (this will pass down to the notifiers) + pull.State = models.PullClosed + + panic("unimplemented") + // s.notifier.NewPullState(ctx, sessDid, pull) +} + +func (s *Service) ReopenPull(ctx context.Context, pull *models.Pull2) error { + panic("unimplemented") +}