johno/beads
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

/{cmd,docs,internal}: support import export for dolt backends

Browse Source
This commit is contained in:
Test
2026-01-21 13:13:24 -08:00
parent 4a0f4abc70
commit b849f598d7
23 changed files with 1837 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
internal/storage/dolt/credentials.go
View File

@@ -263,14 +263,17 @@ func (s *DoltStore) UpdatePeerLastSync(ctx context.Context, name string) error {
// The caller must hold federationEnvMutex.
func setFederationCredentials(username, password string) func() {
if username != "" {
os.Setenv("DOLT_REMOTE_USER", username)
// Best-effort: failures here should not crash the caller.
_ = os.Setenv("DOLT_REMOTE_USER", username)
}
if password != "" {
os.Setenv("DOLT_REMOTE_PASSWORD", password)
// Best-effort: failures here should not crash the caller.
_ = os.Setenv("DOLT_REMOTE_PASSWORD", password)
}
return func() {
os.Unsetenv("DOLT_REMOTE_USER")
os.Unsetenv("DOLT_REMOTE_PASSWORD")
// Best-effort cleanup.
_ = os.Unsetenv("DOLT_REMOTE_USER")
_ = os.Unsetenv("DOLT_REMOTE_PASSWORD")
}
}

29
internal/storage/dolt/events.go
View File

@@ -65,10 +65,26 @@ func (s *DoltStore) GetEvents(ctx context.Context, issueID string, limit int) ([
// AddIssueComment adds a comment to an issue (structured comment)
func (s *DoltStore) AddIssueComment(ctx context.Context, issueID, author, text string) (*types.Comment, error) {
return s.ImportIssueComment(ctx, issueID, author, text, time.Now().UTC())
}
// ImportIssueComment adds a comment during import, preserving the original timestamp.
// This prevents comment timestamp drift across JSONL sync cycles.
func (s *DoltStore) ImportIssueComment(ctx context.Context, issueID, author, text string, createdAt time.Time) (*types.Comment, error) {
// Verify issue exists
var exists bool
if err := s.db.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM issues WHERE id = ?)`, issueID).Scan(&exists); err != nil {
return nil, fmt.Errorf("failed to check issue existence: %w", err)
}
if !exists {
return nil, fmt.Errorf("issue %s not found", issueID)
}
createdAt = createdAt.UTC()
result, err := s.db.ExecContext(ctx, `
INSERT INTO comments (issue_id, author, text, created_at)
VALUES (?, ?, ?, ?)
`, issueID, author, text, time.Now().UTC())
`, issueID, author, text, createdAt)
if err != nil {
return nil, fmt.Errorf("failed to add comment: %w", err)
}
@@ -78,12 +94,21 @@ func (s *DoltStore) AddIssueComment(ctx context.Context, issueID, author, text s
return nil, fmt.Errorf("failed to get comment id: %w", err)
}
// Mark issue dirty for incremental JSONL export
if _, err := s.db.ExecContext(ctx, `
INSERT INTO dirty_issues (issue_id, marked_at)
VALUES (?, ?)
ON DUPLICATE KEY UPDATE marked_at = VALUES(marked_at)
`, issueID, time.Now().UTC()); err != nil {
return nil, fmt.Errorf("failed to mark issue dirty: %w", err)
}
return &types.Comment{
ID: id,
IssueID: issueID,
Author: author,
Text: text,
CreatedAt: time.Now().UTC(),
CreatedAt: createdAt,
}, nil
}

2
internal/storage/dolt/server.go
View File

@@ -103,6 +103,7 @@ func (s *Server) Start(ctx context.Context) error {
}
// Create command
// #nosec G204 -- dolt binary is fixed; args are derived from internal config.
s.cmd = exec.CommandContext(ctx, "dolt", args...)
s.cmd.Dir = s.cfg.DataDir
@@ -272,6 +273,7 @@ func (s *Server) waitForReady(ctx context.Context) error {
// GetRunningServerPID returns the PID of a running server from the PID file, or 0 if not running
func GetRunningServerPID(dataDir string) int {
pidFile := filepath.Join(dataDir, "dolt-server.pid")
// #nosec G304 -- pidFile is derived from internal dataDir.
data, err := os.ReadFile(pidFile)
if err != nil {
return 0

110
internal/storage/dolt/transaction.go
View File

@@ -17,6 +17,12 @@ type doltTransaction struct {
store *DoltStore
}
// CreateIssueImport is the import-friendly issue creation hook.
// Dolt does not enforce prefix validation at the storage layer, so this delegates to CreateIssue.
func (t *doltTransaction) CreateIssueImport(ctx context.Context, issue *types.Issue, actor string, skipPrefixValidation bool) error {
return t.CreateIssue(ctx, issue, actor)
}
// RunInTransaction executes a function within a database transaction
func (s *DoltStore) RunInTransaction(ctx context.Context, fn func(tx storage.Transaction) error) error {
sqlTx, err := s.db.BeginTx(ctx, nil)
@@ -169,6 +175,36 @@ func (t *doltTransaction) AddDependency(ctx context.Context, dep *types.Dependen
return err
}
func (t *doltTransaction) GetDependencyRecords(ctx context.Context, issueID string) ([]*types.Dependency, error) {
rows, err := t.tx.QueryContext(ctx, `
SELECT issue_id, depends_on_id, type, created_at, created_by, metadata, thread_id
FROM dependencies
WHERE issue_id = ?
`, issueID)
if err != nil {
return nil, err
}
defer rows.Close()
var deps []*types.Dependency
for rows.Next() {
var d types.Dependency
var metadata sql.NullString
var threadID sql.NullString
if err := rows.Scan(&d.IssueID, &d.DependsOnID, &d.Type, &d.CreatedAt, &d.CreatedBy, &metadata, &threadID); err != nil {
return nil, err
}
if metadata.Valid {
d.Metadata = metadata.String
}
if threadID.Valid {
d.ThreadID = threadID.String
}
deps = append(deps, &d)
}
return deps, rows.Err()
}
// RemoveDependency removes a dependency within the transaction
func (t *doltTransaction) RemoveDependency(ctx context.Context, issueID, dependsOnID string, actor string) error {
_, err := t.tx.ExecContext(ctx, `
@@ -185,6 +221,23 @@ func (t *doltTransaction) AddLabel(ctx context.Context, issueID, label, actor st
return err
}
func (t *doltTransaction) GetLabels(ctx context.Context, issueID string) ([]string, error) {
rows, err := t.tx.QueryContext(ctx, `SELECT label FROM labels WHERE issue_id = ? ORDER BY label`, issueID)
if err != nil {
return nil, err
}
defer rows.Close()
var labels []string
for rows.Next() {
var l string
if err := rows.Scan(&l); err != nil {
return nil, err
}
labels = append(labels, l)
}
return labels, rows.Err()
}
// RemoveLabel removes a label within the transaction
func (t *doltTransaction) RemoveLabel(ctx context.Context, issueID, label, actor string) error {
_, err := t.tx.ExecContext(ctx, `
@@ -231,6 +284,63 @@ func (t *doltTransaction) GetMetadata(ctx context.Context, key string) (string,
return value, err
}
func (t *doltTransaction) ImportIssueComment(ctx context.Context, issueID, author, text string, createdAt time.Time) (*types.Comment, error) {
// Verify issue exists in tx
iss, err := t.GetIssue(ctx, issueID)
if err != nil {
return nil, err
}
if iss == nil {
return nil, fmt.Errorf("issue %s not found", issueID)
}
createdAt = createdAt.UTC()
res, err := t.tx.ExecContext(ctx, `
INSERT INTO comments (issue_id, author, text, created_at)
VALUES (?, ?, ?, ?)
`, issueID, author, text, createdAt)
if err != nil {
return nil, fmt.Errorf("failed to add comment: %w", err)
}
id, err := res.LastInsertId()
if err != nil {
return nil, fmt.Errorf("failed to get comment id: %w", err)
}
// mark dirty in tx
if _, err := t.tx.ExecContext(ctx, `
INSERT INTO dirty_issues (issue_id, marked_at)
VALUES (?, ?)
ON DUPLICATE KEY UPDATE marked_at = VALUES(marked_at)
`, issueID, time.Now().UTC()); err != nil {
return nil, fmt.Errorf("failed to mark issue dirty: %w", err)
}
return &types.Comment{ID: id, IssueID: issueID, Author: author, Text: text, CreatedAt: createdAt}, nil
}
func (t *doltTransaction) GetIssueComments(ctx context.Context, issueID string) ([]*types.Comment, error) {
rows, err := t.tx.QueryContext(ctx, `
SELECT id, issue_id, author, text, created_at
FROM comments
WHERE issue_id = ?
ORDER BY created_at ASC
`, issueID)
if err != nil {
return nil, err
}
defer rows.Close()
var comments []*types.Comment
for rows.Next() {
var c types.Comment
if err := rows.Scan(&c.ID, &c.IssueID, &c.Author, &c.Text, &c.CreatedAt); err != nil {
return nil, err
}
comments = append(comments, &c)
}
return comments, rows.Err()
}
// AddComment adds a comment within the transaction
func (t *doltTransaction) AddComment(ctx context.Context, issueID, actor, comment string) error {
_, err := t.tx.ExecContext(ctx, `