feat: fail fast when two chatwoot processes mutate the same conversation

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Per-conversation lock for all mutating `conv` verbs (`reply`, `resolve`, `open`, `pending`, `snooze`, `assign`, `unassign`, `label`, `priority`): concurrent mutations of the same conversation from separate terminals now fail fast instead of both running. Locks live in `~/.chatwoot/locks/` and are released automatically by the OS if the process dies.
+
 ### Changed
 
 ### Fixed

go.mod

@@ -1,12 +1,13 @@
 module github.com/chatwoot/cli
 
-go 1.26.4
+go 1.26.5
 
 require (
 	github.com/alecthomas/kong v1.15.0
 	github.com/getkin/kin-openapi v0.140.0
 	github.com/jotaen/kong-completion v0.0.14
 	github.com/zalando/go-keyring v0.2.8
+	golang.org/x/sys v0.46.0
 	golang.org/x/term v0.44.0
 	gopkg.in/yaml.v3 v3.0.1
 )
@@ -25,6 +26,5 @@ require (
 	github.com/posener/complete v1.2.3 // indirect
 	github.com/riywo/loginshell v0.0.0-20200815045211-7d26008be1ab // indirect
 	github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 // indirect
-	golang.org/x/sys v0.46.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 )

internal/cmd/conversation.go

@@ -1,12 +1,14 @@
 package cmd
 
 import (
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
 	"time"
 
 	"github.com/chatwoot/cli/internal/config"
+	"github.com/chatwoot/cli/internal/lock"
 	"github.com/chatwoot/cli/internal/output"
 	"github.com/chatwoot/cli/internal/sdk"
 )
@@ -184,6 +186,25 @@ func (c *ConvMessagesCmd) Run(app *App) error {
 	return nil
 }
 
+// -- per-conversation lock ----------------------------------------------------
+
+// withConvLock runs fn while holding the per-conversation lock, so concurrent
+// chatwoot processes can't mutate the same conversation at once. If another
+// process holds the lock, it fails fast instead of waiting: a queued duplicate
+// would still fire after the holder finishes, which is exactly what the lock
+// exists to prevent.
+func withConvLock(id int, fn func() error) error {
+	lk, err := lock.AcquireConversation(id)
+	if err != nil {
+		if errors.Is(err, lock.ErrLocked) {
+			return fmt.Errorf("conversation %d: another chatwoot command is already running on this conversation; try again in a moment", id)
+		}
+		return err
+	}
+	defer lk.Release()
+	return fn()
+}
+
 // -- reply --------------------------------------------------------------------
 
 type ConvReplyCmd struct {
@@ -193,20 +214,22 @@ type ConvReplyCmd struct {
 }
 
 func (c *ConvReplyCmd) Run(app *App) error {
-	msg, err := app.Client.Messages(c.ID).Create(c.Text, c.Private)
-	if err != nil {
-		return err
-	}
-	if app.Printer.Quiet {
-		fmt.Println(msg.ID)
+	return withConvLock(c.ID, func() error {
+		msg, err := app.Client.Messages(c.ID).Create(c.Text, c.Private)
+		if err != nil {
+			return err
+		}
+		if app.Printer.Quiet {
+			fmt.Println(msg.ID)
+			return nil
+		}
+		kind := "reply"
+		if c.Private {
+			kind = "note"
+		}
+		fmt.Printf("Sent %s on conversation %d (message %d).\n", kind, c.ID, msg.ID)
 		return nil
-	}
-	kind := "reply"
-	if c.Private {
-		kind = "note"
-	}
-	fmt.Printf("Sent %s on conversation %d (message %d).\n", kind, c.ID, msg.ID)
-	return nil
+	})
 }
 
 // -- status verbs: resolve / open / pending / snooze --------------------------
@@ -216,23 +239,29 @@ type ConvResolveCmd struct {
 }
 
 func (c *ConvResolveCmd) Run(app *App) error {
-	return setStatus(app, c.ID, "resolved", nil)
+	return withConvLock(c.ID, func() error {
+		return setStatus(app, c.ID, "resolved", nil)
+	})
 }
 
 type ConvOpenCmd struct {
 	ID int `arg:"" help:"Conversation ID."`
 }
 
 func (c *ConvOpenCmd) Run(app *App) error {
-	return setStatus(app, c.ID, "open", nil)
+	return withConvLock(c.ID, func() error {
+		return setStatus(app, c.ID, "open", nil)
+	})
 }
 
 type ConvPendingCmd struct {
 	ID int `arg:"" help:"Conversation ID."`
 }
 
 func (c *ConvPendingCmd) Run(app *App) error {
-	return setStatus(app, c.ID, "pending", nil)
+	return withConvLock(c.ID, func() error {
+		return setStatus(app, c.ID, "pending", nil)
+	})
 }
 
 type ConvSnoozeCmd struct {
@@ -249,7 +278,9 @@ func (c *ConvSnoozeCmd) Run(app *App) error {
 		}
 		until = &ts
 	}
-	return setStatus(app, c.ID, "snoozed", until)
+	return withConvLock(c.ID, func() error {
+		return setStatus(app, c.ID, "snoozed", until)
+	})
 }
 
 func setStatus(app *App, id int, status string, snoozedUntil *int64) error {
@@ -313,15 +344,21 @@ func (c *ConvAssignCmd) Run(app *App) error {
 	if c.Agent == "" && c.Team == 0 {
 		return fmt.Errorf("--agent or --team required")
 	}
+	// Lock before resolveAgent: its agents/profile lookup is an API call, and
+	// a lock conflict must surface before any request, not as a masked lookup
+	// error or a delayed failure.
 	var agentPtr *int
-	if c.Agent != "" {
-		id, err := resolveAgent(app, c.Agent)
-		if err != nil {
-			return err
+	if err := withConvLock(c.ID, func() error {
+		if c.Agent != "" {
+			id, err := resolveAgent(app, c.Agent)
+			if err != nil {
+				return err
+			}
+			agentPtr = &id
 		}
-		agentPtr = &id
-	}
-	if _, err := app.Client.Conversations().Assign(c.ID, agentPtr, c.Team); err != nil {
+		_, err := app.Client.Conversations().Assign(c.ID, agentPtr, c.Team)
+		return err
+	}); err != nil {
 		return err
 	}
 	if app.Printer.Quiet {
@@ -344,7 +381,9 @@ type ConvUnassignCmd struct {
 }
 
 func (c *ConvUnassignCmd) Run(app *App) error {
-	if err := app.Client.Conversations().Unassign(c.ID); err != nil {
+	if err := withConvLock(c.ID, func() error {
+		return app.Client.Conversations().Unassign(c.ID)
+	}); err != nil {
 		return err
 	}
 	if app.Printer.Quiet {
@@ -371,7 +410,10 @@ func (c *ConvLabelCmd) Run(app *App) error {
 			}
 		}
 	}
-	if _, err := app.Client.Labels(c.ID).Add(flat); err != nil {
+	if err := withConvLock(c.ID, func() error {
+		_, err := app.Client.Labels(c.ID).Add(flat)
+		return err
+	}); err != nil {
 		return err
 	}
 	if app.Printer.Quiet {
@@ -394,7 +436,9 @@ func (c *ConvPriorityCmd) Run(app *App) error {
 	if value == "none" {
 		value = ""
 	}
-	if err := app.Client.Conversations().UpdatePriority(c.ID, value); err != nil {
+	if err := withConvLock(c.ID, func() error {
+		return app.Client.Conversations().UpdatePriority(c.ID, value)
+	}); err != nil {
 		return err
 	}
 	if app.Printer.Quiet {

internal/cmd/conversation_lock_test.go

@@ -0,0 +1,53 @@
+package cmd
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/chatwoot/cli/internal/lock"
+)
+
+// Mutating conversation commands must fail fast when another process holds the
+// per-conversation lock. The App is empty on purpose: the commands must bail
+// out before touching the API client.
+func TestConvMutationsFailWhenConversationLocked(t *testing.T) {
+	home := t.TempDir()
+	t.Setenv("HOME", home)        // unix
+	t.Setenv("USERPROFILE", home) // windows
+
+	const convID = 555
+	lk, err := lock.AcquireConversation(convID)
+	if err != nil {
+		t.Fatalf("AcquireConversation: %v", err)
+	}
+	defer lk.Release()
+
+	app := &App{}
+	cases := []struct {
+		name string
+		run  func() error
+	}{
+		{"reply", func() error { return (&ConvReplyCmd{ID: convID, Text: "hi"}).Run(app) }},
+		{"resolve", func() error { return (&ConvResolveCmd{ID: convID}).Run(app) }},
+		{"open", func() error { return (&ConvOpenCmd{ID: convID}).Run(app) }},
+		{"pending", func() error { return (&ConvPendingCmd{ID: convID}).Run(app) }},
+		{"assign", func() error { return (&ConvAssignCmd{ID: convID, Team: 1}).Run(app) }},
+		// By-name assign: the lock must win before resolveAgent's API lookup.
+		{"assign-by-name", func() error { return (&ConvAssignCmd{ID: convID, Agent: "jane"}).Run(app) }},
+		{"unassign", func() error { return (&ConvUnassignCmd{ID: convID}).Run(app) }},
+		{"snooze", func() error { return (&ConvSnoozeCmd{ID: convID}).Run(app) }},
+		{"label", func() error { return (&ConvLabelCmd{ID: convID, Labels: []string{"vip"}}).Run(app) }},
+		{"priority", func() error { return (&ConvPriorityCmd{ID: convID, Level: "high"}).Run(app) }},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := tc.run()
+			if err == nil {
+				t.Fatal("want error while conversation is locked, got nil")
+			}
+			if !strings.Contains(err.Error(), "already running on this conversation") {
+				t.Errorf("error = %q, want lock-held message", err)
+			}
+		})
+	}
+}

internal/lock/lock.go

@@ -0,0 +1,63 @@
+// Package lock provides per-conversation advisory file locks so that
+// concurrent chatwoot processes on the same machine don't run mutating
+// commands (e.g. reply) against the same conversation at once.
+package lock
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+
+	"github.com/chatwoot/cli/internal/config"
+)
+
+// ErrLocked is returned when another process already holds the lock.
+var ErrLocked = errors.New("conversation is locked by another chatwoot process")
+
+// Lock is a held per-conversation lock. Release it with Release.
+// The OS releases it automatically if the process exits or crashes.
+type Lock struct {
+	f *os.File
+}
+
+// AcquireConversation takes an exclusive non-blocking lock for the given
+// conversation ID. It returns ErrLocked (wrapped) if another process holds it.
+func AcquireConversation(id int) (*Lock, error) {
+	cfgDir, err := config.ConfigDir()
+	if err != nil {
+		return nil, err
+	}
+	return acquireAt(filepath.Join(cfgDir, "locks"), id)
+}
+
+func acquireAt(dir string, id int) (*Lock, error) {
+	if err := os.MkdirAll(dir, 0o700); err != nil {
+		return nil, err
+	}
+	path := filepath.Join(dir, fmt.Sprintf("conv-%d.lock", id))
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
+	if err != nil {
+		return nil, err
+	}
+	if err := tryLock(f); err != nil {
+		_ = f.Close()
+		return nil, fmt.Errorf("conversation %d: %w", id, err)
+	}
+	// Best-effort PID marker for debugging; the flock is the real lock.
+	_ = f.Truncate(0)
+	_, _ = f.WriteString(strconv.Itoa(os.Getpid()) + "\n")
+	return &Lock{f: f}, nil
+}
+
+// Release unlocks and closes the lock file. The file itself is left in place:
+// unlinking it would race with other processes locking the same path.
+func (l *Lock) Release() {
+	if l == nil || l.f == nil {
+		return
+	}
+	_ = unlock(l.f)
+	_ = l.f.Close()
+	l.f = nil
+}

internal/lock/lock_test.go

@@ -0,0 +1,45 @@
+package lock
+
+import (
+	"errors"
+	"path/filepath"
+	"testing"
+)
+
+func TestAcquireConflictAndRelease(t *testing.T) {
+	dir := filepath.Join(t.TempDir(), "locks")
+
+	l1, err := acquireAt(dir, 42)
+	if err != nil {
+		t.Fatalf("first acquire: %v", err)
+	}
+
+	if _, err := acquireAt(dir, 42); !errors.Is(err, ErrLocked) {
+		t.Fatalf("second acquire: want ErrLocked, got %v", err)
+	}
+
+	// A different conversation is unaffected.
+	l2, err := acquireAt(dir, 43)
+	if err != nil {
+		t.Fatalf("acquire other conversation: %v", err)
+	}
+	l2.Release()
+
+	l1.Release()
+	l3, err := acquireAt(dir, 42)
+	if err != nil {
+		t.Fatalf("reacquire after release: %v", err)
+	}
+	l3.Release()
+}
+
+func TestReleaseIsIdempotent(t *testing.T) {
+	l, err := acquireAt(filepath.Join(t.TempDir(), "locks"), 1)
+	if err != nil {
+		t.Fatalf("acquire: %v", err)
+	}
+	l.Release()
+	l.Release() // must not panic
+	var nilLock *Lock
+	nilLock.Release() // nil receiver must be safe
+}