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

Merge pull request #1240 from coffeegoddd/db/import-export

Browse Source 
Enable full-fidelity JSONL import/export for Dolt backend
This commit is contained in:
Steve Yegge
2026-01-21 16:40:03 -08:00
committed by GitHub
parent cb46db603d 01fa2d9f3f
commit caf4c368c4
27 changed files with 2011 additions and 347 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/importer/backend_agnostic_import_test.go
+114
View File
@@ -0,0 +1,114 @@
//go:build !integration
// +build !integration
package importer
import (
"context"
"testing"
"time"
"github.com/steveyegge/beads/internal/storage/memory"
"github.com/steveyegge/beads/internal/types"
)
func TestImportIssues_BackendAgnostic_DepsLabelsCommentsTombstone(t *testing.T) {
ctx := context.Background()
store := memory.New("")
if err := store.SetConfig(ctx, "issue_prefix", "test"); err != nil {
t.Fatalf("set issue_prefix: %v", err)
}
commentTS := time.Date(2020, 1, 2, 3, 4, 5, 6, time.UTC)
deletedTS := time.Date(2021, 2, 3, 4, 5, 6, 7, time.UTC)
issueA := &types.Issue{
ID: "test-1",
Title: "Issue A",
IssueType: types.TypeTask,
Status: types.StatusOpen,
Priority: 2,
Labels: []string{"urgent"},
Dependencies: []*types.Dependency{
{IssueID: "test-1", DependsOnID: "test-2", Type: types.DepBlocks},
},
Comments: []*types.Comment{
{Author: "tester", Text: "hello", CreatedAt: commentTS},
},
}
issueB := &types.Issue{
ID: "test-2",
Title: "Issue B",
IssueType: types.TypeTask,
Status: types.StatusTombstone,
Priority: 4,
DeletedAt: &deletedTS,
DeletedBy: "tester",
DeleteReason: "bye",
OriginalType: string(types.TypeTask),
Description: "tombstone",
ContentHash: "",
Dependencies: nil,
Labels: nil,
Comments: nil,
Assignee: "",
Owner: "",
CreatedBy: "",
SourceSystem: "",
ExternalRef: nil,
ClosedAt: nil,
CompactedAt: nil,
DeferUntil: nil,
LastActivity: nil,
QualityScore: nil,
Validations: nil,
BondedFrom: nil,
Waiters: nil,
}
res, err := ImportIssues(ctx, "", store, []*types.Issue{issueA, issueB}, Options{OrphanHandling: OrphanAllow})
if err != nil {
t.Fatalf("ImportIssues: %v", err)
}
if res.Created != 2 {
t.Fatalf("expected Created=2, got %d", res.Created)
}
labels, err := store.GetLabels(ctx, "test-1")
if err != nil {
t.Fatalf("GetLabels: %v", err)
}
if len(labels) != 1 || labels[0] != "urgent" {
t.Fatalf("expected labels [urgent], got %v", labels)
}
deps, err := store.GetDependencyRecords(ctx, "test-1")
if err != nil {
t.Fatalf("GetDependencyRecords: %v", err)
}
if len(deps) != 1 || deps[0].DependsOnID != "test-2" || deps[0].Type != types.DepBlocks {
t.Fatalf("expected dependency test-1 blocks test-2, got %#v", deps)
}
comments, err := store.GetIssueComments(ctx, "test-1")
if err != nil {
t.Fatalf("GetIssueComments: %v", err)
}
if len(comments) != 1 {
t.Fatalf("expected 1 comment, got %d", len(comments))
}
if !comments[0].CreatedAt.Equal(commentTS) {
t.Fatalf("expected comment timestamp preserved (%s), got %s", commentTS.Format(time.RFC3339Nano), comments[0].CreatedAt.Format(time.RFC3339Nano))
}
b, err := store.GetIssue(ctx, "test-2")
if err != nil {
t.Fatalf("GetIssue: %v", err)
}
if b.Status != types.StatusTombstone {
t.Fatalf("expected tombstone status, got %q", b.Status)
}
if b.DeletedAt == nil || !b.DeletedAt.Equal(deletedTS) {
t.Fatalf("expected DeletedAt preserved (%s), got %#v", deletedTS.Format(time.RFC3339Nano), b.DeletedAt)
}
}
internal/importer/importer.go
+708 -145
View File
File diff suppressed because it is too large Load Diff
internal/importer/importer_test.go
+92 -140
View File
@@ -153,10 +153,10 @@ func TestFieldComparator_StringConversion(t *testing.T) {
fc := newFieldComparator()
tests := []struct {
name string
value interface{}
wantStr string
wantOk bool
name string
value interface{}
wantStr string
wantOk bool
}{
{"string", "hello", "hello", true},
{"string pointer", stringPtr("world"), "world", true},
@@ -436,67 +436,67 @@ func TestRenameImportedIssuePrefixes(t *testing.T) {
func TestReplaceBoundaryAware(t *testing.T) {
tests := []struct {
name string
text string
oldID string
newID string
want string
name string
text string
oldID string
newID string
want string
}{
{
name: "simple replacement",
text: "See old-1 for details",
oldID: "old-1",
newID: "new-1",
want: "See new-1 for details",
name: "simple replacement",
text: "See old-1 for details",
oldID: "old-1",
newID: "new-1",
want: "See new-1 for details",
},
{
name: "multiple occurrences",
text: "old-1 and old-1 again",
oldID: "old-1",
newID: "new-1",
want: "new-1 and new-1 again",
name: "multiple occurrences",
text: "old-1 and old-1 again",
oldID: "old-1",
newID: "new-1",
want: "new-1 and new-1 again",
},
{
name: "no match substring prefix",
text: "old-10 should not match",
oldID: "old-1",
newID: "new-1",
want: "old-10 should not match",
name: "no match substring prefix",
text: "old-10 should not match",
oldID: "old-1",
newID: "new-1",
want: "old-10 should not match",
},
{
name: "match at end of longer ID",
text: "should not match old-1 at end",
oldID: "old-1",
newID: "new-1",
want: "should not match new-1 at end",
name: "match at end of longer ID",
text: "should not match old-1 at end",
oldID: "old-1",
newID: "new-1",
want: "should not match new-1 at end",
},
{
name: "boundary at start",
text: "old-1 starts here",
oldID: "old-1",
newID: "new-1",
want: "new-1 starts here",
name: "boundary at start",
text: "old-1 starts here",
oldID: "old-1",
newID: "new-1",
want: "new-1 starts here",
},
{
name: "boundary at end",
text: "ends with old-1",
oldID: "old-1",
newID: "new-1",
want: "ends with new-1",
name: "boundary at end",
text: "ends with old-1",
oldID: "old-1",
newID: "new-1",
want: "ends with new-1",
},
{
name: "boundary punctuation",
text: "See (old-1) and [old-1] or {old-1}",
oldID: "old-1",
newID: "new-1",
want: "See (new-1) and [new-1] or {new-1}",
name: "boundary punctuation",
text: "See (old-1) and [old-1] or {old-1}",
oldID: "old-1",
newID: "new-1",
want: "See (new-1) and [new-1] or {new-1}",
},
{
name: "no occurrence",
text: "No match here",
oldID: "old-1",
newID: "new-1",
want: "No match here",
name: "no occurrence",
text: "No match here",
oldID: "old-1",
newID: "new-1",
want: "No match here",
},
}
@@ -547,9 +547,9 @@ func TestIsValidIDSuffix(t *testing.T) {
{"abc.1.2.3", true},
{"1.5", true},
// Invalid suffixes
{"", false}, // Empty string
{"A3F8", false}, // Uppercase not allowed
{"@#$!", false}, // Special characters not allowed
{"", false}, // Empty string
{"A3F8", false}, // Uppercase not allowed
{"@#$!", false}, // Special characters not allowed
{"abc-def", false}, // Hyphens not allowed in suffix
}
@@ -569,7 +569,7 @@ func stringPtr(s string) *string {
func TestImportIssues_Basic(t *testing.T) {
ctx := context.Background()
// Create temp database
tmpDB := t.TempDir() + "/test.db"
store, err := sqlite.New(context.Background(), tmpDB)
@@ -577,12 +577,12 @@ func TestImportIssues_Basic(t *testing.T) {
t.Fatalf("Failed to create store: %v", err)
}
defer store.Close()
// Set config prefix
if err := store.SetConfig(ctx, "issue_prefix", "test"); err != nil {
t.Fatalf("Failed to set prefix: %v", err)
}
// Import single issue
issues := []*types.Issue{
{
@@ -594,16 +594,16 @@ func TestImportIssues_Basic(t *testing.T) {
IssueType: types.TypeTask,
},
}
result, err := ImportIssues(ctx, tmpDB, store, issues, Options{})
if err != nil {
t.Fatalf("Import failed: %v", err)
}
if result.Created != 1 {
t.Errorf("Expected 1 created, got %d", result.Created)
}
// Verify issue was created
retrieved, err := store.GetIssue(ctx, "test-abc123")
if err != nil {
@@ -616,18 +616,18 @@ func TestImportIssues_Basic(t *testing.T) {
func TestImportIssues_Update(t *testing.T) {
ctx := context.Background()
tmpDB := t.TempDir() + "/test.db"
store, err := sqlite.New(context.Background(), tmpDB)
if err != nil {
t.Fatalf("Failed to create store: %v", err)
}
defer store.Close()
if err := store.SetConfig(ctx, "issue_prefix", "test"); err != nil {
t.Fatalf("Failed to set prefix: %v", err)
}
// Create initial issue
issue1 := &types.Issue{
ID: "test-abc123",
@@ -638,12 +638,12 @@ func TestImportIssues_Update(t *testing.T) {
IssueType: types.TypeTask,
}
issue1.ContentHash = issue1.ComputeContentHash()
err = store.CreateIssue(ctx, issue1, "test")
if err != nil {
t.Fatalf("Failed to create initial issue: %v", err)
}
// Import updated version with newer timestamp
issue2 := &types.Issue{
ID: "test-abc123",
@@ -656,18 +656,18 @@ func TestImportIssues_Update(t *testing.T) {
UpdatedAt: time.Now().Add(time.Hour), // Newer than issue1
}
issue2.ContentHash = issue2.ComputeContentHash()
result, err := ImportIssues(ctx, tmpDB, store, []*types.Issue{issue2}, Options{})
if err != nil {
t.Fatalf("Import failed: %v", err)
}
// The importer detects this as both a collision (1) and then upserts it (creates=1)
// Total updates = collision count + actual upserts
if result.Updated == 0 && result.Created == 0 {
t.Error("Expected some updates or creates")
}
// Verify update
retrieved, err := store.GetIssue(ctx, "test-abc123")
if err != nil {
@@ -680,18 +680,18 @@ func TestImportIssues_Update(t *testing.T) {
func TestImportIssues_DryRun(t *testing.T) {
ctx := context.Background()
tmpDB := t.TempDir() + "/test.db"
store, err := sqlite.New(context.Background(), tmpDB)
if err != nil {
t.Fatalf("Failed to create store: %v", err)
}
defer store.Close()
if err := store.SetConfig(ctx, "issue_prefix", "test"); err != nil {
t.Fatalf("Failed to set prefix: %v", err)
}
issues := []*types.Issue{
{
ID: "test-abc123",
@@ -701,13 +701,13 @@ func TestImportIssues_DryRun(t *testing.T) {
IssueType: types.TypeTask,
},
}
// Dry run returns early when no collisions, so it reports what would be created
result, err := ImportIssues(ctx, tmpDB, store, issues, Options{DryRun: true})
if err != nil {
t.Fatalf("Import failed: %v", err)
}
// Should report that 1 issue would be created
if result.Created != 1 {
t.Errorf("Expected 1 would be created in dry run, got %d", result.Created)
@@ -716,18 +716,18 @@ func TestImportIssues_DryRun(t *testing.T) {
func TestImportIssues_Dependencies(t *testing.T) {
ctx := context.Background()
tmpDB := t.TempDir() + "/test.db"
store, err := sqlite.New(context.Background(), tmpDB)
if err != nil {
t.Fatalf("Failed to create store: %v", err)
}
defer store.Close()
if err := store.SetConfig(ctx, "issue_prefix", "test"); err != nil {
t.Fatalf("Failed to set prefix: %v", err)
}
issues := []*types.Issue{
{
ID: "test-abc123",
@@ -747,16 +747,16 @@ func TestImportIssues_Dependencies(t *testing.T) {
IssueType: types.TypeTask,
},
}
result, err := ImportIssues(ctx, tmpDB, store, issues, Options{})
if err != nil {
t.Fatalf("Import failed: %v", err)
}
if result.Created != 2 {
t.Errorf("Expected 2 created, got %d", result.Created)
}
// Verify dependency was created
deps, err := store.GetDependencies(ctx, "test-abc123")
if err != nil {
@@ -769,18 +769,18 @@ func TestImportIssues_Dependencies(t *testing.T) {
func TestImportIssues_Labels(t *testing.T) {
ctx := context.Background()
tmpDB := t.TempDir() + "/test.db"
store, err := sqlite.New(context.Background(), tmpDB)
if err != nil {
t.Fatalf("Failed to create store: %v", err)
}
defer store.Close()
if err := store.SetConfig(ctx, "issue_prefix", "test"); err != nil {
t.Fatalf("Failed to set prefix: %v", err)
}
issues := []*types.Issue{
{
ID: "test-abc123",
@@ -791,16 +791,16 @@ func TestImportIssues_Labels(t *testing.T) {
Labels: []string{"bug", "critical"},
},
}
result, err := ImportIssues(ctx, tmpDB, store, issues, Options{})
if err != nil {
t.Fatalf("Import failed: %v", err)
}
if result.Created != 1 {
t.Errorf("Expected 1 created, got %d", result.Created)
}
// Verify labels were created
retrieved, err := store.GetIssue(ctx, "test-abc123")
if err != nil {
@@ -811,61 +811,13 @@ func TestImportIssues_Labels(t *testing.T) {
}
}
func TestGetOrCreateStore_ExistingStore(t *testing.T) {
ctx := context.Background()
tmpDB := t.TempDir() + "/test.db"
store, err := sqlite.New(context.Background(), tmpDB)
if err != nil {
t.Fatalf("Failed to create store: %v", err)
}
defer store.Close()
result, needClose, err := getOrCreateStore(ctx, tmpDB, store)
if err != nil {
t.Fatalf("Expected no error, got: %v", err)
}
if needClose {
t.Error("Expected needClose=false for existing store")
}
if result != store {
t.Error("Expected same store instance")
}
}
// NOTE: getOrCreateStore was removed; importer now requires an initialized store.
func TestGetOrCreateStore_NewStore(t *testing.T) {
func TestImportIssues_RequiresStore(t *testing.T) {
ctx := context.Background()
tmpDB := t.TempDir() + "/test.db"
// Create initial database
initStore, err := sqlite.New(context.Background(), tmpDB)
if err != nil {
t.Fatalf("Failed to create store: %v", err)
}
initStore.Close()
// Test creating new connection
result, needClose, err := getOrCreateStore(ctx, tmpDB, nil)
if err != nil {
t.Fatalf("Expected no error, got: %v", err)
}
defer result.Close()
if !needClose {
t.Error("Expected needClose=true for new store")
}
if result == nil {
t.Error("Expected non-nil store")
}
}
func TestGetOrCreateStore_EmptyPath(t *testing.T) {
ctx := context.Background()
_, _, err := getOrCreateStore(ctx, "", nil)
_, err := ImportIssues(ctx, "", nil, []*types.Issue{}, Options{})
if err == nil {
t.Error("Expected error for empty database path")
t.Fatal("expected error when store is nil")
}
}
@@ -891,7 +843,7 @@ func TestGetPrefixList(t *testing.T) {
want: []string{},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := GetPrefixList(tt.prefixes)
@@ -1010,7 +962,7 @@ func TestValidateNoDuplicateExternalRefs(t *testing.T) {
func TestConcurrentExternalRefImports(t *testing.T) {
t.Skip("TODO(bd-gpe7): Test hangs due to database deadlock - needs investigation")
t.Run("sequential imports with same external_ref are detected as updates", func(t *testing.T) {
store, err := sqlite.New(context.Background(), ":memory:")
if err != nil {
@@ -1024,7 +976,7 @@ func TestConcurrentExternalRefImports(t *testing.T) {
}
externalRef := "JIRA-100"
issue1 := &types.Issue{
ID: "bd-1",
Title: "First import",
@@ -1182,7 +1134,7 @@ func TestImportOrphanSkip_CountMismatch(t *testing.T) {
UpdatedAt: now,
},
{
ID: "test-orphan.1", // Child of non-existent parent
ID: "test-orphan.1", // Child of non-existent parent
Title: "Orphaned Child",
Status: types.StatusOpen,
Priority: 2,
@@ -1203,7 +1155,7 @@ func TestImportOrphanSkip_CountMismatch(t *testing.T) {
// Import with OrphanSkip mode - parent doesn't exist
result, err := ImportIssues(ctx, "", store, issues, Options{
OrphanHandling: sqlite.OrphanSkip,
OrphanHandling: OrphanSkip,
SkipPrefixValidation: true, // Allow explicit IDs during import
})
if err != nil {
internal/storage/dolt/credentials.go
+7 -4
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")
}
}
internal/storage/dolt/events.go
+27 -2
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
}
internal/storage/dolt/procattr_unix.go
+15
View File
@@ -0,0 +1,15 @@
//go:build !windows
// +build !windows
package dolt
import (
"os/exec"
"syscall"
)
func setDoltServerSysProcAttr(cmd *exec.Cmd) {
cmd.SysProcAttr = &syscall.SysProcAttr{
Setpgid: true,
}
}
internal/storage/dolt/procattr_windows.go
+12
View File
@@ -0,0 +1,12 @@
//go:build windows
// +build windows
package dolt
import "os/exec"
// Windows does not support Setpgid; leave default process attributes.
func setDoltServerSysProcAttr(cmd *exec.Cmd) {
// no-op
_ = cmd
}
internal/storage/dolt/process_unix.go
+32
View File
@@ -0,0 +1,32 @@
//go:build !windows
// +build !windows
package dolt
import (
"os"
"strings"
"syscall"
)
func processMayBeAlive(p *os.Process) bool {
// Signal 0 checks for existence without sending a real signal.
if err := p.Signal(syscall.Signal(0)); err != nil {
return false
}
return true
}
func terminateProcess(p *os.Process) error {
if p == nil {
return nil
}
if err := p.Signal(syscall.SIGTERM); err != nil {
// Process may already be dead; treat as success.
if strings.Contains(err.Error(), "process already finished") {
return nil
}
return err
}
return nil
}
internal/storage/dolt/process_windows.go
+21
View File
@@ -0,0 +1,21 @@
//go:build windows
// +build windows
package dolt
import "os"
func processMayBeAlive(p *os.Process) bool {
// Windows doesn't support Unix-style signal(0) checks. Treat as "unknown/alive"
// and let connection attempts / wait timeouts determine readiness.
_ = p
return true
}
func terminateProcess(p *os.Process) error {
if p == nil {
return nil
}
// Best-effort: Windows doesn't have SIGTERM semantics; kill the process.
return p.Kill()
}
internal/storage/dolt/server.go
+13 -27
View File
@@ -15,7 +15,6 @@ import (
"strconv"
"strings"
"sync"
"syscall"
"time"
)
@@ -103,13 +102,12 @@ 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
// Set up process group for clean shutdown
s.cmd.SysProcAttr = &syscall.SysProcAttr{
Setpgid: true,
}
// Set up process group for clean shutdown (Unix-only; no-op on Windows).
setDoltServerSysProcAttr(s.cmd)
// Set up logging
if s.cfg.LogFile != "" {
@@ -162,13 +160,8 @@ func (s *Server) Stop() error {
return nil
}
// Try graceful shutdown first (SIGTERM)
if err := s.cmd.Process.Signal(syscall.SIGTERM); err != nil {
// Process may already be dead
if !strings.Contains(err.Error(), "process already finished") {
return fmt.Errorf("failed to send SIGTERM: %w", err)
}
}
// Best-effort graceful shutdown (platform-specific).
_ = terminateProcess(s.cmd.Process)
// Wait for graceful shutdown with timeout
done := make(chan error, 1)
@@ -249,11 +242,9 @@ func (s *Server) waitForReady(ctx context.Context) error {
default:
}
// Check if process is still alive using signal 0
if s.cmd.Process != nil {
if err := s.cmd.Process.Signal(syscall.Signal(0)); err != nil {
return fmt.Errorf("server process exited unexpectedly")
}
// Best-effort: if we can tell the process is dead, fail fast.
if s.cmd.Process != nil && !processMayBeAlive(s.cmd.Process) {
return fmt.Errorf("server process exited unexpectedly")
}
// Try to connect
@@ -272,6 +263,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
@@ -288,9 +280,8 @@ func GetRunningServerPID(dataDir string) int {
return 0
}
// On Unix, FindProcess always succeeds, so we need to check if it's alive
if err := process.Signal(syscall.Signal(0)); err != nil {
// Process is not running
// Best-effort liveness check (platform-specific).
if !processMayBeAlive(process) {
_ = os.Remove(pidFile)
return 0
}
@@ -305,13 +296,8 @@ func StopServerByPID(pid int) error {
return err
}
// Try graceful shutdown first
if err := process.Signal(syscall.SIGTERM); err != nil {
if !strings.Contains(err.Error(), "process already finished") {
return err
}
return nil
}
// Best-effort graceful shutdown (platform-specific).
_ = terminateProcess(process)
// Wait for graceful shutdown
done := make(chan struct{})
internal/storage/dolt/transaction.go
+110
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, `
internal/storage/memory/memory.go
+18
View File
@@ -1458,6 +1458,24 @@ func (m *MemoryStorage) AddIssueComment(ctx context.Context, issueID, author, te
return comment, nil
}
func (m *MemoryStorage) ImportIssueComment(ctx context.Context, issueID, author, text string, createdAt time.Time) (*types.Comment, error) {
m.mu.Lock()
defer m.mu.Unlock()
comment := &types.Comment{
ID: int64(len(m.comments[issueID]) + 1),
IssueID: issueID,
Author: author,
Text: text,
CreatedAt: createdAt,
}
m.comments[issueID] = append(m.comments[issueID], comment)
m.dirty[issueID] = true
return comment, nil
}
func (m *MemoryStorage) GetIssueComments(ctx context.Context, issueID string) ([]*types.Comment, error) {
m.mu.RLock()
defer m.mu.RUnlock()
internal/storage/sqlite/comments.go
+4 -2
View File
@@ -3,6 +3,7 @@ package sqlite
import (
"context"
"fmt"
"time"
"github.com/steveyegge/beads/internal/types"
)
@@ -56,7 +57,7 @@ func (s *SQLiteStorage) AddIssueComment(ctx context.Context, issueID, author, te
// Unlike AddIssueComment which uses CURRENT_TIMESTAMP, this method uses the provided
// createdAt time from the JSONL file. This prevents timestamp drift during sync cycles.
// GH#735: Comment created_at timestamps were being overwritten with current time during import.
func (s *SQLiteStorage) ImportIssueComment(ctx context.Context, issueID, author, text string, createdAt string) (*types.Comment, error) {
func (s *SQLiteStorage) ImportIssueComment(ctx context.Context, issueID, author, text string, createdAt time.Time) (*types.Comment, error) {
// Verify issue exists
var exists bool
err := s.db.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM issues WHERE id = ?)`, issueID).Scan(&exists)
@@ -68,10 +69,11 @@ func (s *SQLiteStorage) ImportIssueComment(ctx context.Context, issueID, author,
}
// Insert comment with provided timestamp
createdAtStr := createdAt.UTC().Format(time.RFC3339Nano)
result, err := s.db.ExecContext(ctx, `
INSERT INTO comments (issue_id, author, text, created_at)
VALUES (?, ?, ?, ?)
`, issueID, author, text, createdAt)
`, issueID, author, text, createdAtStr)
if err != nil {
return nil, fmt.Errorf("failed to insert comment: %w", err)
}
internal/storage/sqlite/import_tx.go
+115
View File
@@ -0,0 +1,115 @@
package sqlite
import (
"context"
"database/sql"
"fmt"
"time"
"github.com/steveyegge/beads/internal/types"
)
// CreateIssueImport creates an issue inside an existing sqlite transaction, optionally skipping
// prefix validation. This is used by JSONL import to support multi-repo mode (GH#686).
func (t *sqliteTxStorage) CreateIssueImport(ctx context.Context, issue *types.Issue, actor string, skipPrefixValidation bool) error {
// Fetch custom statuses and types for validation
customStatuses, err := t.GetCustomStatuses(ctx)
if err != nil {
return fmt.Errorf("failed to get custom statuses: %w", err)
}
customTypes, err := t.GetCustomTypes(ctx)
if err != nil {
return fmt.Errorf("failed to get custom types: %w", err)
}
// Set timestamps
now := time.Now()
if issue.CreatedAt.IsZero() {
issue.CreatedAt = now
}
if issue.UpdatedAt.IsZero() {
issue.UpdatedAt = now
}
// Defensive fix for closed_at invariant
if issue.Status == types.StatusClosed && issue.ClosedAt == nil {
maxTime := issue.CreatedAt
if issue.UpdatedAt.After(maxTime) {
maxTime = issue.UpdatedAt
}
closedAt := maxTime.Add(time.Second)
issue.ClosedAt = &closedAt
}
// Defensive fix for tombstone invariant
if issue.Status == types.StatusTombstone && issue.DeletedAt == nil {
maxTime := issue.CreatedAt
if issue.UpdatedAt.After(maxTime) {
maxTime = issue.UpdatedAt
}
deletedAt := maxTime.Add(time.Second)
issue.DeletedAt = &deletedAt
}
// Validate issue before creating
if err := issue.ValidateWithCustom(customStatuses, customTypes); err != nil {
return fmt.Errorf("validation failed: %w", err)
}
// Compute content hash
if issue.ContentHash == "" {
issue.ContentHash = issue.ComputeContentHash()
}
// Get configured prefix for validation and ID generation behavior
var configPrefix string
err = t.conn.QueryRowContext(ctx, `SELECT value FROM config WHERE key = ?`, "issue_prefix").Scan(&configPrefix)
if err == sql.ErrNoRows || configPrefix == "" {
return fmt.Errorf("database not initialized: issue_prefix config is missing (run 'bd init --prefix <prefix>' first)")
} else if err != nil {
return fmt.Errorf("failed to get config: %w", err)
}
prefix := configPrefix
if issue.IDPrefix != "" {
prefix = configPrefix + "-" + issue.IDPrefix
}
if issue.ID == "" {
// Import path expects IDs, but be defensive and generate if missing.
generatedID, err := GenerateIssueID(ctx, t.conn, prefix, issue, actor)
if err != nil {
return fmt.Errorf("failed to generate issue ID: %w", err)
}
issue.ID = generatedID
} else if !skipPrefixValidation {
if err := ValidateIssueIDPrefix(issue.ID, prefix); err != nil {
return fmt.Errorf("failed to validate issue ID prefix: %w", err)
}
}
// Ensure parent exists for hierarchical IDs (importer should have ensured / resurrected).
if isHierarchical, parentID := IsHierarchicalID(issue.ID); isHierarchical {
var parentCount int
if err := t.conn.QueryRowContext(ctx, `SELECT COUNT(*) FROM issues WHERE id = ?`, parentID).Scan(&parentCount); err != nil {
return fmt.Errorf("failed to check parent existence: %w", err)
}
if parentCount == 0 {
return fmt.Errorf("parent issue %s does not exist", parentID)
}
}
// Insert issue (strict)
if err := insertIssueStrict(ctx, t.conn, issue); err != nil {
return fmt.Errorf("failed to insert issue: %w", err)
}
// Record event
if err := recordCreatedEvent(ctx, t.conn, issue, actor); err != nil {
return fmt.Errorf("failed to record creation event: %w", err)
}
// Mark dirty
if err := markDirty(ctx, t.conn, issue.ID); err != nil {
return fmt.Errorf("failed to mark issue dirty: %w", err)
}
return nil
}
internal/storage/sqlite/transaction.go
+98
View File
@@ -824,6 +824,37 @@ func (t *sqliteTxStorage) AddDependency(ctx context.Context, dep *types.Dependen
return nil
}
// GetDependencyRecords retrieves dependency records for an issue within the transaction.
func (t *sqliteTxStorage) GetDependencyRecords(ctx context.Context, issueID string) ([]*types.Dependency, error) {
rows, err := t.conn.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, fmt.Errorf("failed to query dependencies: %w", err)
}
defer func() { _ = 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, fmt.Errorf("failed to scan dependency: %w", 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 *sqliteTxStorage) RemoveDependency(ctx context.Context, issueID, dependsOnID string, actor string) error {
// First, check what type of dependency is being removed
@@ -916,6 +947,11 @@ func (t *sqliteTxStorage) AddLabel(ctx context.Context, issueID, label, actor st
return nil
}
// GetLabels retrieves labels for an issue within the transaction.
func (t *sqliteTxStorage) GetLabels(ctx context.Context, issueID string) ([]string, error) {
return t.getLabels(ctx, issueID)
}
// RemoveLabel removes a label from an issue within the transaction.
func (t *sqliteTxStorage) RemoveLabel(ctx context.Context, issueID, label, actor string) error {
result, err := t.conn.ExecContext(ctx, `
@@ -1063,6 +1099,68 @@ func (t *sqliteTxStorage) AddComment(ctx context.Context, issueID, actor, commen
return nil
}
// ImportIssueComment adds a structured comment during import, preserving the original timestamp.
func (t *sqliteTxStorage) ImportIssueComment(ctx context.Context, issueID, author, text string, createdAt time.Time) (*types.Comment, error) {
// Verify issue exists
existing, err := t.GetIssue(ctx, issueID)
if err != nil {
return nil, fmt.Errorf("failed to check issue existence: %w", err)
}
if existing == nil {
return nil, fmt.Errorf("issue %s not found", issueID)
}
createdAtStr := createdAt.UTC().Format(time.RFC3339Nano)
res, err := t.conn.ExecContext(ctx, `
INSERT INTO comments (issue_id, author, text, created_at)
VALUES (?, ?, ?, ?)
`, issueID, author, text, createdAtStr)
if err != nil {
return nil, fmt.Errorf("failed to insert comment: %w", err)
}
commentID, err := res.LastInsertId()
if err != nil {
return nil, fmt.Errorf("failed to get comment ID: %w", err)
}
// Mark issue dirty
if err := markDirty(ctx, t.conn, issueID); err != nil {
return nil, fmt.Errorf("failed to mark issue dirty: %w", err)
}
return &types.Comment{
ID: commentID,
IssueID: issueID,
Author: author,
Text: text,
CreatedAt: createdAt.UTC(),
}, nil
}
// GetIssueComments retrieves structured comments for an issue within the transaction.
func (t *sqliteTxStorage) GetIssueComments(ctx context.Context, issueID string) ([]*types.Comment, error) {
rows, err := t.conn.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, fmt.Errorf("failed to query comments: %w", err)
}
defer func() { _ = 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, fmt.Errorf("failed to scan comment: %w", err)
}
comments = append(comments, &c)
}
return comments, rows.Err()
}
// SearchIssues finds issues matching query and filters within the transaction.
// This enables read-your-writes semantics for searching within a transaction.
func (t *sqliteTxStorage) SearchIssues(ctx context.Context, query string, filter types.IssueFilter) ([]*types.Issue, error) {
internal/storage/storage.go
+8
View File
@@ -4,6 +4,7 @@ package storage
import (
"context"
"database/sql"
"time"
"github.com/steveyegge/beads/internal/types"
)
@@ -58,10 +59,12 @@ type Transaction interface {
// Dependency operations
AddDependency(ctx context.Context, dep *types.Dependency, actor string) error
RemoveDependency(ctx context.Context, issueID, dependsOnID string, actor string) error
GetDependencyRecords(ctx context.Context, issueID string) ([]*types.Dependency, error)
// Label operations
AddLabel(ctx context.Context, issueID, label, actor string) error
RemoveLabel(ctx context.Context, issueID, label, actor string) error
GetLabels(ctx context.Context, issueID string) ([]string, error)
// Config operations (for atomic config + issue workflows)
SetConfig(ctx context.Context, key, value string) error
@@ -73,6 +76,8 @@ type Transaction interface {
// Comment operations
AddComment(ctx context.Context, issueID, actor, comment string) error
ImportIssueComment(ctx context.Context, issueID, author, text string, createdAt time.Time) (*types.Comment, error)
GetIssueComments(ctx context.Context, issueID string) ([]*types.Comment, error)
}
// Storage defines the interface for issue storage backends
@@ -121,6 +126,9 @@ type Storage interface {
// Comments
AddIssueComment(ctx context.Context, issueID, author, text string) (*types.Comment, error)
// ImportIssueComment adds a comment while preserving the original timestamp.
// Used during JSONL import to avoid timestamp drift across sync cycles.
ImportIssueComment(ctx context.Context, issueID, author, text string, createdAt time.Time) (*types.Comment, error)
GetIssueComments(ctx context.Context, issueID string) ([]*types.Comment, error)
GetCommentsForIssues(ctx context.Context, issueIDs []string) (map[string][]*types.Comment, error)
internal/storage/storage_test.go
+16
View File
@@ -5,6 +5,7 @@ import (
"context"
"database/sql"
"testing"
"time"
"github.com/steveyegge/beads/internal/types"
)
@@ -119,6 +120,9 @@ func (m *mockStorage) GetEvents(ctx context.Context, issueID string, limit int)
func (m *mockStorage) AddIssueComment(ctx context.Context, issueID, author, text string) (*types.Comment, error) {
return nil, nil
}
func (m *mockStorage) ImportIssueComment(ctx context.Context, issueID, author, text string, createdAt time.Time) (*types.Comment, error) {
return nil, nil
}
func (m *mockStorage) GetIssueComments(ctx context.Context, issueID string) ([]*types.Comment, error) {
return nil, nil
}
@@ -237,12 +241,18 @@ func (m *mockTransaction) AddDependency(ctx context.Context, dep *types.Dependen
func (m *mockTransaction) RemoveDependency(ctx context.Context, issueID, dependsOnID string, actor string) error {
return nil
}
func (m *mockTransaction) GetDependencyRecords(ctx context.Context, issueID string) ([]*types.Dependency, error) {
return nil, nil
}
func (m *mockTransaction) AddLabel(ctx context.Context, issueID, label, actor string) error {
return nil
}
func (m *mockTransaction) RemoveLabel(ctx context.Context, issueID, label, actor string) error {
return nil
}
func (m *mockTransaction) GetLabels(ctx context.Context, issueID string) ([]string, error) {
return nil, nil
}
func (m *mockTransaction) SetConfig(ctx context.Context, key, value string) error {
return nil
}
@@ -258,6 +268,12 @@ func (m *mockTransaction) GetMetadata(ctx context.Context, key string) (string,
func (m *mockTransaction) AddComment(ctx context.Context, issueID, actor, comment string) error {
return nil
}
func (m *mockTransaction) ImportIssueComment(ctx context.Context, issueID, author, text string, createdAt time.Time) (*types.Comment, error) {
return nil, nil
}
func (m *mockTransaction) GetIssueComments(ctx context.Context, issueID string) ([]*types.Comment, error) {
return nil, nil
}
// TestConfig verifies the Config struct has expected fields.
func TestConfig(t *testing.T) {