diff --git a/internal/http/contact_merge_handlers.go b/internal/http/contact_merge_handlers.go index 1e960b4b..501c8500 100644 --- a/internal/http/contact_merge_handlers.go +++ b/internal/http/contact_merge_handlers.go @@ -50,6 +50,7 @@ func (h *ChannelInstancesHandler) handleMergeContacts(w http.ResponseWriter, r * } var targetID uuid.UUID + var targetUserID string // tenant_user.user_id string for context file migration if hasTU { // Link to existing tenant_user — verify same tenant. @@ -63,6 +64,7 @@ func (h *ChannelInstancesHandler) handleMergeContacts(w http.ResponseWriter, r * return } targetID = tu.ID + targetUserID = tu.UserID } else { // Create new tenant_user. userID := body.CreateUser.UserID @@ -82,6 +84,7 @@ func (h *ChannelInstancesHandler) handleMergeContacts(w http.ResponseWriter, r * return } targetID = tu.ID + targetUserID = tu.UserID } if err := h.contactStore.MergeContacts(r.Context(), body.ContactIDs, targetID); err != nil { @@ -90,6 +93,9 @@ func (h *ChannelInstancesHandler) handleMergeContacts(w http.ResponseWriter, r * return } + // Migrate user_context_files from old sender_ids to new tenant_user_id. + h.migrateContextFilesOnMerge(r.Context(), body.ContactIDs, targetUserID) + emitAudit(h.msgBus, r, "contacts.merged", "tenant_user", targetID.String()) writeJSON(w, http.StatusOK, map[string]any{ "merged_id": targetID, @@ -182,6 +188,30 @@ func (h *ChannelInstancesHandler) handleListTenantUsers(w http.ResponseWriter, r writeJSON(w, http.StatusOK, map[string]any{"users": users}) } +// migrateContextFilesOnMerge moves user_context_files from old sender_ids to the new tenant_user_id. +// Best-effort: log errors but don't fail the merge. +func (h *ChannelInstancesHandler) migrateContextFilesOnMerge(ctx context.Context, contactIDs []uuid.UUID, newUserID string) { + // Batch-fetch sender_ids from merged contacts in one query. + oldUserIDs, err := h.contactStore.GetSenderIDsByContactIDs(ctx, contactIDs) + if err != nil { + slog.Warn("contacts.merge.get_sender_ids", "error", err) + return + } + // Filter out the target user_id itself (no self-migration needed). + filtered := oldUserIDs[:0] + for _, id := range oldUserIDs { + if id != newUserID { + filtered = append(filtered, id) + } + } + if len(filtered) == 0 { + return + } + if err := h.agentStore.MigrateUserDataOnMerge(ctx, filtered, newUserID); err != nil { + slog.Warn("contacts.merge.migrate_context_files", "error", err, "old_ids", filtered, "new_id", newUserID) + } +} + // deriveUserIDFromContacts returns the first contact's username or sender_id as fallback user_id. func (h *ChannelInstancesHandler) deriveUserIDFromContacts(ctx context.Context, contactIDs []uuid.UUID) string { if len(contactIDs) == 0 { diff --git a/internal/store/agent_store.go b/internal/store/agent_store.go index a0201871..ab26a22e 100644 --- a/internal/store/agent_store.go +++ b/internal/store/agent_store.go @@ -505,6 +505,10 @@ type AgentContextStore interface { GetUserContextFiles(ctx context.Context, agentID uuid.UUID, userID string) ([]UserContextFileData, error) SetUserContextFile(ctx context.Context, agentID uuid.UUID, userID, fileName, content string) error DeleteUserContextFile(ctx context.Context, agentID uuid.UUID, userID, fileName string) error + // MigrateUserDataOnMerge moves per-user data from oldUserIDs to newUserID when contacts are merged. + // Covers: user_context_files, user_agent_overrides, user_agent_profiles, memory_documents/chunks. + // On conflict, keeps the newest by updated_at. Best-effort per table. + MigrateUserDataOnMerge(ctx context.Context, oldUserIDs []string, newUserID string) error GetUserOverride(ctx context.Context, agentID uuid.UUID, userID string) (*UserAgentOverrideData, error) SetUserOverride(ctx context.Context, override *UserAgentOverrideData) error } diff --git a/internal/store/contact_store.go b/internal/store/contact_store.go index fb9bcbe4..d3f55c80 100644 --- a/internal/store/contact_store.go +++ b/internal/store/contact_store.go @@ -53,6 +53,9 @@ type ContactStore interface { // GetContactByID returns a single contact by primary key. Tenant-scoped via context. GetContactByID(ctx context.Context, id uuid.UUID) (*ChannelContact, error) + // GetSenderIDsByContactIDs returns sender_id strings for the given contact UUIDs in one query. + GetSenderIDsByContactIDs(ctx context.Context, contactIDs []uuid.UUID) ([]string, error) + // MergeContacts sets merged_id = tenantUserID on all given contact IDs, // linking them to a tenant_users identity. Tenant-scoped via context. MergeContacts(ctx context.Context, contactIDs []uuid.UUID, tenantUserID uuid.UUID) error diff --git a/internal/store/pg/agents_context.go b/internal/store/pg/agents_context.go index 26d4bbae..362a63c2 100644 --- a/internal/store/pg/agents_context.go +++ b/internal/store/pg/agents_context.go @@ -5,7 +5,10 @@ import ( "database/sql" "encoding/json" "errors" + "fmt" + "log/slog" "path/filepath" + "strings" "time" "github.com/google/uuid" @@ -123,6 +126,89 @@ func (s *PGAgentStore) DeleteUserContextFile(ctx context.Context, agentID uuid.U return err } +func (s *PGAgentStore) MigrateUserDataOnMerge(ctx context.Context, oldUserIDs []string, newUserID string) error { + if len(oldUserIDs) == 0 { + return nil + } + tClause, tArgs, _, err := scopeClause(ctx, len(oldUserIDs)+2) + if err != nil { + return err + } + + placeholders := make([]string, len(oldUserIDs)) + baseArgs := make([]any, 0, len(oldUserIDs)+1+len(tArgs)) + for i, id := range oldUserIDs { + placeholders[i] = fmt.Sprintf("$%d", i+1) + baseArgs = append(baseArgs, id) + } + inClause := strings.Join(placeholders, ",") + newP := fmt.Sprintf("$%d", len(oldUserIDs)+1) + baseArgs = append(baseArgs, newUserID) + baseArgs = append(baseArgs, tArgs...) + + // Args for delete (no newUserID param needed). + delArgs := make([]any, len(oldUserIDs)) + copy(delArgs, baseArgs[:len(oldUserIDs)]) + delArgs = append(delArgs, tArgs...) + + // Helper: migrate + delete for one table. DO NOTHING on conflict — + // existing tenant user data always wins (canonical identity). + migrate := func(insertQ, deleteQ string) { + if _, err := s.db.ExecContext(ctx, insertQ, baseArgs...); err != nil { + slog.Warn("merge.migrate", "error", err) + } + if _, err := s.db.ExecContext(ctx, deleteQ, delArgs...); err != nil { + slog.Warn("merge.cleanup", "error", err) + } + } + + // 1. user_context_files: UNIQUE(agent_id, user_id, file_name) + migrate( + fmt.Sprintf(`INSERT INTO user_context_files (id, agent_id, user_id, file_name, content, updated_at, tenant_id) + SELECT gen_random_uuid(), agent_id, %s, file_name, content, updated_at, tenant_id + FROM user_context_files WHERE user_id IN (%s)%s + ON CONFLICT (agent_id, user_id, file_name) DO NOTHING`, newP, inClause, tClause), + fmt.Sprintf(`DELETE FROM user_context_files WHERE user_id IN (%s)%s`, inClause, tClause), + ) + + // 2. user_agent_overrides: UNIQUE(agent_id, user_id) + migrate( + fmt.Sprintf(`INSERT INTO user_agent_overrides (id, agent_id, user_id, provider, model, settings, created_at, updated_at, tenant_id) + SELECT gen_random_uuid(), agent_id, %s, provider, model, settings, created_at, updated_at, tenant_id + FROM user_agent_overrides WHERE user_id IN (%s)%s + ON CONFLICT (agent_id, user_id) DO NOTHING`, newP, inClause, tClause), + fmt.Sprintf(`DELETE FROM user_agent_overrides WHERE user_id IN (%s)%s`, inClause, tClause), + ) + + // 3. user_agent_profiles: PK(agent_id, user_id) + migrate( + fmt.Sprintf(`INSERT INTO user_agent_profiles (agent_id, user_id, workspace, first_seen_at, last_seen_at, metadata, tenant_id) + SELECT agent_id, %s, workspace, first_seen_at, last_seen_at, metadata, tenant_id + FROM user_agent_profiles WHERE user_id IN (%s)%s + ON CONFLICT (agent_id, user_id) DO NOTHING`, newP, inClause, tClause), + fmt.Sprintf(`DELETE FROM user_agent_profiles WHERE user_id IN (%s)%s`, inClause, tClause), + ) + + // 4. memory_documents: UNIQUE(agent_id, COALESCE(user_id,''), path) + migrate( + fmt.Sprintf(`INSERT INTO memory_documents (id, agent_id, user_id, path, content, hash, updated_at, created_at, tenant_id) + SELECT gen_random_uuid(), agent_id, %s, path, content, hash, updated_at, created_at, tenant_id + FROM memory_documents WHERE user_id IN (%s)%s + ON CONFLICT (agent_id, COALESCE(user_id,''), path) DO NOTHING`, newP, inClause, tClause), + fmt.Sprintf(`DELETE FROM memory_documents WHERE user_id IN (%s)%s`, inClause, tClause), + ) + + // 5. memory_chunks: FK on document_id — cascade from memory_documents delete handles this. + // But orphan chunks (where document was already migrated) need cleanup. + // Simply re-point remaining chunks whose document still has old user_id. + repoint := fmt.Sprintf(`UPDATE memory_chunks SET user_id = %s WHERE user_id IN (%s)%s`, newP, inClause, tClause) + if _, err := s.db.ExecContext(ctx, repoint, baseArgs...); err != nil { + slog.Warn("merge.migrate_chunks", "error", err) + } + + return nil +} + // --- User-Agent Profiles --- func (s *PGAgentStore) GetOrCreateUserProfile(ctx context.Context, agentID uuid.UUID, userID, workspace, channel string) (bool, string, error) { diff --git a/internal/store/pg/channel_contacts.go b/internal/store/pg/channel_contacts.go index cc49ee2c..f052d9c8 100644 --- a/internal/store/pg/channel_contacts.go +++ b/internal/store/pg/channel_contacts.go @@ -194,6 +194,36 @@ func (s *PGContactStore) GetContactByID(ctx context.Context, id uuid.UUID) (*sto return &c, nil } +func (s *PGContactStore) GetSenderIDsByContactIDs(ctx context.Context, contactIDs []uuid.UUID) ([]string, error) { + if len(contactIDs) == 0 { + return nil, nil + } + tid := store.TenantIDFromContext(ctx) + placeholders := make([]string, len(contactIDs)) + args := make([]any, len(contactIDs)+1) + for i, id := range contactIDs { + placeholders[i] = fmt.Sprintf("$%d", i+1) + args[i] = id + } + args[len(contactIDs)] = tid + q := fmt.Sprintf("SELECT sender_id FROM channel_contacts WHERE id IN (%s) AND tenant_id = $%d", + strings.Join(placeholders, ","), len(contactIDs)+1) + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var result []string + for rows.Next() { + var sid string + if err := rows.Scan(&sid); err != nil { + return nil, err + } + result = append(result, sid) + } + return result, rows.Err() +} + func (s *PGContactStore) MergeContacts(ctx context.Context, contactIDs []uuid.UUID, tenantUserID uuid.UUID) error { if len(contactIDs) == 0 { return nil diff --git a/internal/store/sqlitestore/agents_context.go b/internal/store/sqlitestore/agents_context.go index 7d0b6c43..8fb42639 100644 --- a/internal/store/sqlitestore/agents_context.go +++ b/internal/store/sqlitestore/agents_context.go @@ -7,7 +7,10 @@ import ( "database/sql" "encoding/json" "errors" + "fmt" + "log/slog" "path/filepath" + "strings" "time" "github.com/google/uuid" @@ -133,6 +136,86 @@ func (s *SQLiteAgentStore) DeleteUserContextFile(ctx context.Context, agentID uu return err } +func (s *SQLiteAgentStore) MigrateUserDataOnMerge(ctx context.Context, oldUserIDs []string, newUserID string) error { + if len(oldUserIDs) == 0 { + return nil + } + tClause, tArgs, err := scopeClause(ctx) + if err != nil { + return err + } + + placeholders := make([]string, len(oldUserIDs)) + baseArgs := make([]any, 0, len(oldUserIDs)+1+len(tArgs)) + for i, id := range oldUserIDs { + placeholders[i] = "?" + baseArgs = append(baseArgs, id) + } + inClause := strings.Join(placeholders, ",") + baseArgs = append(baseArgs, newUserID) + baseArgs = append(baseArgs, tArgs...) + + delArgs := make([]any, len(oldUserIDs)) + copy(delArgs, baseArgs[:len(oldUserIDs)]) + delArgs = append(delArgs, tArgs...) + + uuid := `lower(hex(randomblob(4))||'-'||hex(randomblob(2))||'-'||hex(randomblob(2))||'-'||hex(randomblob(2))||'-'||hex(randomblob(6)))` + + // DO NOTHING on conflict — existing tenant user data always wins. + migrate := func(insertQ, deleteQ string) { + if _, err := s.db.ExecContext(ctx, insertQ, baseArgs...); err != nil { + slog.Warn("merge.migrate", "error", err) + } + if _, err := s.db.ExecContext(ctx, deleteQ, delArgs...); err != nil { + slog.Warn("merge.cleanup", "error", err) + } + } + + // 1. user_context_files + migrate( + fmt.Sprintf(`INSERT INTO user_context_files (id, agent_id, user_id, file_name, content, updated_at, tenant_id) + SELECT %s, agent_id, ?, file_name, content, updated_at, tenant_id + FROM user_context_files WHERE user_id IN (%s)%s + ON CONFLICT (agent_id, user_id, file_name) DO NOTHING`, uuid, inClause, tClause), + fmt.Sprintf(`DELETE FROM user_context_files WHERE user_id IN (%s)%s`, inClause, tClause), + ) + + // 2. user_agent_overrides + migrate( + fmt.Sprintf(`INSERT INTO user_agent_overrides (id, agent_id, user_id, provider, model, settings, created_at, updated_at, tenant_id) + SELECT %s, agent_id, ?, provider, model, settings, created_at, updated_at, tenant_id + FROM user_agent_overrides WHERE user_id IN (%s)%s + ON CONFLICT (agent_id, user_id) DO NOTHING`, uuid, inClause, tClause), + fmt.Sprintf(`DELETE FROM user_agent_overrides WHERE user_id IN (%s)%s`, inClause, tClause), + ) + + // 3. user_agent_profiles + migrate( + fmt.Sprintf(`INSERT INTO user_agent_profiles (agent_id, user_id, workspace, first_seen_at, last_seen_at, metadata, tenant_id) + SELECT agent_id, ?, workspace, first_seen_at, last_seen_at, metadata, tenant_id + FROM user_agent_profiles WHERE user_id IN (%s)%s + ON CONFLICT (agent_id, user_id) DO NOTHING`, inClause, tClause), + fmt.Sprintf(`DELETE FROM user_agent_profiles WHERE user_id IN (%s)%s`, inClause, tClause), + ) + + // 4. memory_documents + migrate( + fmt.Sprintf(`INSERT INTO memory_documents (id, agent_id, user_id, path, content, hash, updated_at, created_at, tenant_id) + SELECT %s, agent_id, ?, path, content, hash, updated_at, created_at, tenant_id + FROM memory_documents WHERE user_id IN (%s)%s + ON CONFLICT (agent_id, COALESCE(user_id,''), path) DO NOTHING`, uuid, inClause, tClause), + fmt.Sprintf(`DELETE FROM memory_documents WHERE user_id IN (%s)%s`, inClause, tClause), + ) + + // 5. memory_chunks: re-point remaining chunks. + repoint := fmt.Sprintf(`UPDATE memory_chunks SET user_id = ? WHERE user_id IN (%s)%s`, inClause, tClause) + if _, err := s.db.ExecContext(ctx, repoint, baseArgs...); err != nil { + slog.Warn("merge.migrate_chunks", "error", err) + } + + return nil +} + // --- User-Agent Profiles --- // GetOrCreateUserProfile upserts a user profile and returns (isNew, effectiveWorkspace, error). diff --git a/internal/store/sqlitestore/contacts.go b/internal/store/sqlitestore/contacts.go index 40d0e7d2..66f3d1fb 100644 --- a/internal/store/sqlitestore/contacts.go +++ b/internal/store/sqlitestore/contacts.go @@ -191,6 +191,36 @@ func (s *SQLiteContactStore) GetContactByID(ctx context.Context, id uuid.UUID) ( return &c, nil } +func (s *SQLiteContactStore) GetSenderIDsByContactIDs(ctx context.Context, contactIDs []uuid.UUID) ([]string, error) { + if len(contactIDs) == 0 { + return nil, nil + } + tid := store.TenantIDFromContext(ctx) + placeholders := make([]string, len(contactIDs)) + args := make([]any, len(contactIDs)+1) + for i, id := range contactIDs { + placeholders[i] = "?" + args[i] = id + } + args[len(contactIDs)] = tid + q := fmt.Sprintf("SELECT sender_id FROM channel_contacts WHERE id IN (%s) AND tenant_id = ?", + strings.Join(placeholders, ",")) + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var result []string + for rows.Next() { + var sid string + if err := rows.Scan(&sid); err != nil { + return nil, err + } + result = append(result, sid) + } + return result, rows.Err() +} + func (s *SQLiteContactStore) MergeContacts(ctx context.Context, contactIDs []uuid.UUID, tenantUserID uuid.UUID) error { if len(contactIDs) == 0 { return nil diff --git a/ui/web/src/components/shared/multi-user-picker.tsx b/ui/web/src/components/shared/multi-user-picker.tsx new file mode 100644 index 00000000..e3dce092 --- /dev/null +++ b/ui/web/src/components/shared/multi-user-picker.tsx @@ -0,0 +1,76 @@ +import { useState } from "react"; +import { X } from "lucide-react"; +import { Badge } from "@/components/ui/badge"; +import { UserPickerCombobox } from "./user-picker-combobox"; + +interface MultiUserPickerProps { + value: string[]; + onChange: (values: string[]) => void; + placeholder?: string; + /** Filter by source: "contact" | "tenant_user" | undefined (both). */ + source?: "contact" | "tenant_user"; + /** Filter contacts by peer_kind. */ + peerKind?: "direct" | "group"; + /** Allow typing custom values not in the list. Default true. */ + allowCustom?: boolean; + /** Render dropdown into a portal container (useful inside dialogs). */ + portalContainer?: React.RefObject; +} + +/** + * Multi-select user picker: search + select → badge list. + * Wraps UserPickerCombobox for multi-value fields (allow_from, deny lists). + * + * Uses `onSelect` (fires on dropdown click / custom commit) instead of + * `onChange` (fires on every keystroke) to avoid adding partial text. + */ +export function MultiUserPicker({ + value, + onChange, + placeholder, + source, + peerKind, + allowCustom = true, + portalContainer, +}: MultiUserPickerProps) { + const [inputValue, setInputValue] = useState(""); + + const handleCommit = (val: string) => { + const trimmed = val.trim(); + if (trimmed && !value.includes(trimmed)) { + onChange([...value, trimmed]); + } + setInputValue(""); + }; + + return ( +
+ + {value.length > 0 && ( +
+ {value.map((id) => ( + + {id} + + + ))} +
+ )} +
+ ); +} diff --git a/ui/web/src/components/shared/user-picker-combobox.tsx b/ui/web/src/components/shared/user-picker-combobox.tsx index ea4fbbd4..729ad7ce 100644 --- a/ui/web/src/components/shared/user-picker-combobox.tsx +++ b/ui/web/src/components/shared/user-picker-combobox.tsx @@ -4,6 +4,8 @@ import { useUserPicker } from "@/hooks/use-user-picker"; interface UserPickerComboboxProps { value: string; onChange: (value: string) => void; + /** Fires only on dropdown item click or custom value commit (not keystrokes). */ + onSelect?: (value: string) => void; placeholder?: string; className?: string; /** Filter contacts by peer_kind: "direct" | "group" | undefined (all). */ @@ -31,6 +33,7 @@ interface UserPickerComboboxProps { export function UserPickerCombobox({ value, onChange, + onSelect, placeholder, className, peerKind, @@ -44,6 +47,7 @@ export function UserPickerCombobox({ void; + /** Fires only on dropdown item click or custom value commit (not on keystrokes). + * Use this for multi-select wrappers that need to distinguish typing from selection. */ + onSelect?: (value: string) => void; options: ComboboxOption[]; placeholder?: string; className?: string; @@ -25,6 +28,7 @@ interface ComboboxProps { export function Combobox({ value, onChange, + onSelect, options, placeholder, className, @@ -75,28 +79,62 @@ export function Combobox({ // eslint-disable-next-line react-hooks/exhaustive-deps }, [portalContainer, open]); - // Compute dropdown position — always use fixed positioning for portal rendering + // Compute dropdown position — flip above input when near viewport bottom. + // When flipped up, use `bottom` anchoring so the dropdown grows upward + // from the input edge (filtering reduces items without creating a gap). React.useLayoutEffect(() => { if (!open || !containerRef.current) return; const inputRect = containerRef.current.getBoundingClientRect(); + const DROP_H = 256; // max-h-60 ≈ 240px + border/padding + const GAP = 4; + const spaceBelow = window.innerHeight - inputRect.bottom; + const flipUp = spaceBelow < DROP_H && inputRect.top > DROP_H; + if (resolvedPortal) { const portalRect = resolvedPortal.getBoundingClientRect(); const scrollTop = resolvedPortal.scrollTop || 0; const scrollLeft = resolvedPortal.scrollLeft || 0; const left = inputRect.left - portalRect.left + scrollLeft; const maxWidth = portalRect.width - (inputRect.left - portalRect.left); + const portalRelativeTop = inputRect.top - portalRect.top + scrollTop; + const portalFlipUp = spaceBelow < DROP_H && portalRelativeTop > DROP_H; + if (portalFlipUp) { + // Anchor bottom edge to input top — dropdown grows upward. + const portalH = resolvedPortal.scrollHeight || portalRect.height; + setDropdownStyle({ + position: "absolute", + bottom: portalH - portalRelativeTop + GAP, + left, + width: inputRect.width, + maxWidth, + maxHeight: DROP_H, + zIndex: 50, + }); + } else { + setDropdownStyle({ + position: "absolute", + top: inputRect.bottom - portalRect.top + scrollTop + GAP, + left, + width: inputRect.width, + maxWidth, + zIndex: 50, + }); + } + } else if (flipUp) { + // Fixed: anchor bottom edge to input top — dropdown grows upward. + const bottomFromViewport = window.innerHeight - inputRect.top; setDropdownStyle({ - position: "absolute", - top: inputRect.bottom - portalRect.top + scrollTop + 4, - left, + position: "fixed", + bottom: bottomFromViewport + GAP, + left: inputRect.left, width: inputRect.width, - maxWidth, - zIndex: 50, + maxHeight: DROP_H, + zIndex: 9999, }); } else { setDropdownStyle({ position: "fixed", - top: inputRect.bottom + 4, + top: inputRect.bottom + GAP, left: inputRect.left, width: inputRect.width, zIndex: 9999, @@ -125,6 +163,7 @@ export function Combobox({ const handleSelect = (val: string) => { onChange(val); + onSelect?.(val); const match = options.find((o) => o.value === val); setSearch(match?.label || val); setOpen(false); diff --git a/ui/web/src/pages/channels/channel-fields.tsx b/ui/web/src/pages/channels/channel-fields.tsx index 27bb06a9..2db4e618 100644 --- a/ui/web/src/pages/channels/channel-fields.tsx +++ b/ui/web/src/pages/channels/channel-fields.tsx @@ -3,6 +3,7 @@ import { Input } from "@/components/ui/input"; import { Label } from "@/components/ui/label"; import { Switch } from "@/components/ui/switch"; import { Textarea } from "@/components/ui/textarea"; +import { MultiUserPicker } from "@/components/shared/multi-user-picker"; import { Select, SelectContent, @@ -274,16 +275,10 @@ function FieldRenderer({ return (
-