diff --git a/README.md b/README.md index e3005c38..3c701f02 100644 --- a/README.md +++ b/README.md @@ -277,3 +277,45 @@ Debugging a live install (failing sends, re-pairing, the two-data-dir gotcha, si ## License MIT + +### Complete Google Contacts directory (optional, legacy read mode) + +Google Messages supplies a limited contact suggestion list. Names in existing +threads often arrive independently through conversation updates; a successful +Google Messages connection does not prove the whole address book was imported. + +To keep a complete directory, configure an authenticated Google Contacts MCP +server exposing the `contacts_list` tool with Google People API `connections`, +`nextPageToken`, and `totalItems` (or `totalPeople`) fields. For example, +[google-contacts-mcp](https://github.com/domdomegg/google-contacts-mcp) supports +this contract. Authorize that connector for the same Google account as the phone. +OpenMessage calls only its read-only listing tool; OAuth stays in the connector. + +- `OPENMESSAGES_GOOGLE_CONTACTS_MCP_URL`: opt-in Streamable HTTP endpoint, such as + `http://127.0.0.1:3230/mcp`. HTTP requires a literal loopback IP; remote endpoints + require HTTPS. Redirects and credentials/query strings in the URL are rejected. +- `OPENMESSAGES_GOOGLE_CONTACTS_MCP_TOKEN_FILE`: optional private (0600) file + containing a connector bearer token, if required. Do not put a token in the URL. +- `OPENMESSAGES_CONTACTS_REFRESH_INTERVAL`: refresh interval, default `5m`, + minimum `1m`, maximum `24h`. + +The daemon imports immediately and on the interval, even when the phone is +unavailable. All pages must arrive and agree with the reported total before a +single transaction replaces the previous directory. Failures preserve that last +complete snapshot. All returned phones, emails and organizations are retained; +full dialable phone numbers become compose suggestions. Email-only records remain +in the directory but do not create SMS routes or empty conversations. + +For existing SMS threads, unambiguous numbers replace blank/numeric names, and +names supplied by the previous directory follow renames/deletions. Shared numbers +are left unresolved. Custom titles and group titles are preserved, as are message +history, phone-side contact IDs, read state and favorites. Later raw-number +conversation snapshots also consult the directory. + +`GET /api/status` includes `contact_sync` with the source, running state, last +attempt/success times, people/phone counts and refresh errors. `complete` describes +the last successfully imported snapshot; inspect `last_error` and the success +age as well. `POST /api/contacts/sync` performs a manual refresh. With no connector +configured, the existing Google Messages suggestion fetch remains available and +is not reported as a complete directory. Demo, client-only and v2-primary modes +do not run the full-directory worker; v2 directory integration is future work. diff --git a/cmd/serve.go b/cmd/serve.go index 983d0f11..416ae09a 100644 --- a/cmd/serve.go +++ b/cmd/serve.go @@ -161,6 +161,7 @@ func RunServe(logger zerolog.Logger, args ...string) error { if err != nil { return fmt.Errorf("init app: %w", err) } + a.ContactDirectoryDisabled = isDemo || v2Primary || !transports defer a.Close() interactiveTerminal := term.IsTerminal(int(os.Stdin.Fd())) @@ -659,6 +660,9 @@ func RunServe(logger zerolog.Logger, args ...string) error { // same store double-sends every due message. if transports { startLegacyScheduler(v2Primary, a.StartScheduler) + if !isDemo && !v2Primary { + a.StartContactDirectorySync() + } } v2Options := v2SendWebOptions(stack, v2Send) @@ -717,6 +721,7 @@ func RunServe(logger zerolog.Logger, args ...string) error { BackfillStatus: func() any { return a.GetBackfillProgress() }, BackfillPhone: a.BackfillConversationByPhone, SyncGoogleContacts: a.SyncGoogleContacts, + ContactSyncStatus: func() any { return a.GetContactSyncStatus() }, }) } else { httpHandler = web.ProtectLocalControl(controlAuth.Handler(mcpHTTPHandler)) diff --git a/internal/app/app.go b/internal/app/app.go index d71f37bd..498e77c6 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -112,6 +112,10 @@ func (p *BackfillProgress) snapshot() BackfillSnapshot { } type App struct { + // Set before starting transports; directory writes currently target legacy reads. + ContactDirectoryDisabled bool + + contactDirectory contactDirectoryState clientMu sync.RWMutex Client *client.Client googleGeneration *GoogleGeneration @@ -870,6 +874,7 @@ func (a *App) GetBackfillProgress() BackfillSnapshot { } func (a *App) Close() { + a.stopContactDirectorySync() a.StopGoogleAvatarSync() if cli := a.GetClient(); cli != nil { cli.GM.Disconnect() diff --git a/internal/app/contact_directory.go b/internal/app/contact_directory.go new file mode 100644 index 00000000..51af5ea4 --- /dev/null +++ b/internal/app/contact_directory.go @@ -0,0 +1,168 @@ +package app + +import ( + "context" + "errors" + "os" + "strings" + "sync" + "time" + + "github.com/maxghenis/openmessage/internal/contactsync" +) + +type ContactSyncStatus struct { + Source string `json:"source"` + Running bool `json:"running"` + Complete bool `json:"complete"` + LastAttemptMS int64 `json:"last_attempt_ms"` + LastSuccessMS int64 `json:"last_success_ms"` + People int `json:"people"` + PhoneEntries int `json:"phone_entries"` + ThreadsUpdated int `json:"threads_updated"` + RefreshIntervalSeconds int `json:"refresh_interval_seconds"` + LastError string `json:"last_error,omitempty"` +} +type contactDirectoryState struct { + mu sync.Mutex + workMu sync.Mutex + wg sync.WaitGroup + cancel context.CancelFunc + ctx context.Context + closed bool + status ContactSyncStatus +} + +func googleContactsMCPURL() string { + return strings.TrimSpace(os.Getenv("OPENMESSAGES_GOOGLE_CONTACTS_MCP_URL")) +} +func contactRefreshInterval() time.Duration { + if v, err := time.ParseDuration(strings.TrimSpace(os.Getenv("OPENMESSAGES_CONTACTS_REFRESH_INTERVAL"))); err == nil && v >= time.Minute && v <= 24*time.Hour { + return v + } + return 5 * time.Minute +} +func (a *App) GetContactSyncStatus() ContactSyncStatus { + d := &a.contactDirectory + d.mu.Lock() + defer d.mu.Unlock() + s := d.status + if a.ContactDirectoryDisabled { + s.Source = "disabled" + return s + } + if s.Source == "" { + s.Source = "google_messages_suggestions" + if googleContactsMCPURL() != "" { + s.Source = "google_contacts_mcp" + } + } + if s.Source == "google_contacts_mcp" { + s.RefreshIntervalSeconds = int(contactRefreshInterval() / time.Second) + } + return s +} + +// StartContactDirectorySync is daemon-only and independent of phone connectivity. +// Reconnects cannot create additional workers. Close cancels and joins requests. +func (a *App) StartContactDirectorySync() { + if a == nil || a.ContactDirectoryDisabled || googleContactsMCPURL() == "" { + return + } + d := &a.contactDirectory + d.mu.Lock() + if d.closed || d.cancel != nil { + d.mu.Unlock() + return + } + ctx, cancel := context.WithCancel(context.Background()) + d.ctx = ctx + d.cancel = cancel + d.wg.Add(1) + d.mu.Unlock() + go func() { + defer d.wg.Done() + runContactRefreshLoop(ctx, contactRefreshInterval(), func() { + if _, err := a.syncContactDirectory(ctx); err != nil && ctx.Err() == nil { + a.Logger.Warn().Msg("Google Contacts directory refresh failed; check contact_sync status") + } + }) + }() +} +func runContactRefreshLoop(ctx context.Context, interval time.Duration, refresh func()) { + refresh() + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + refresh() + } + } +} +func (a *App) stopContactDirectorySync() { + d := &a.contactDirectory + d.mu.Lock() + d.closed = true + if d.cancel != nil { + d.cancel() + } + d.mu.Unlock() + d.wg.Wait() + // A manual refresh can be in flight even before the periodic worker starts. + d.workMu.Lock() + d.workMu.Unlock() +} +func (a *App) syncContactDirectory(parent context.Context) (int, error) { + if a.ContactDirectoryDisabled { + return 0, errors.New("contact directory sync requires a legacy daemon") + } + d := &a.contactDirectory + d.workMu.Lock() + defer d.workMu.Unlock() + d.mu.Lock() + if d.closed { + d.mu.Unlock() + return 0, errors.New("contact sync is stopped") + } + // Tie manual requests to daemon shutdown as well as their timeout. + if d.ctx != nil { + parent = d.ctx + } + d.status.Source = "google_contacts_mcp" + d.status.Running = true + d.status.LastAttemptMS = time.Now().UnixMilli() + d.mu.Unlock() + ctx, cancel := context.WithTimeout(parent, 90*time.Second) + defer cancel() + people, err := contactsync.Fetch(ctx, googleContactsMCPURL(), strings.TrimSpace(os.Getenv("OPENMESSAGES_GOOGLE_CONTACTS_MCP_TOKEN_FILE"))) + phones, changed := 0, 0 + if err == nil { + if ctx.Err() != nil { + err = ctx.Err() + } else { + phones, changed, err = a.Store.ReplaceContactDirectory(people) + } + } + d.mu.Lock() + d.status.Running = false + if err != nil { + d.status.LastError = "Contact directory refresh failed; verify connector availability and authorization, then retry" + } else { + d.status.LastSuccessMS = time.Now().UnixMilli() + d.status.Complete = true + d.status.People = len(people) + d.status.PhoneEntries = phones + d.status.ThreadsUpdated = changed + d.status.LastError = "" + } + d.mu.Unlock() + if err != nil { + return 0, err + } + a.Logger.Info().Int("people", len(people)).Int("phone_entries", phones).Int("threads_updated", changed).Msg("Google Contacts directory refreshed") + a.emitConversationsChange() + return phones, nil +} diff --git a/internal/app/contact_directory_test.go b/internal/app/contact_directory_test.go new file mode 100644 index 00000000..2f6ad5bb --- /dev/null +++ b/internal/app/contact_directory_test.go @@ -0,0 +1,73 @@ +package app + +import ( + "context" + "sync/atomic" + "testing" + "time" +) + +func TestContactRefreshLoopRepeatsAndCancels(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + var calls atomic.Int32 + go func() { + defer close(done) + runContactRefreshLoop(ctx, time.Millisecond, func() { + if calls.Add(1) == 3 { + cancel() + } + }) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("worker did not stop") + } + if calls.Load() < 3 { + t.Fatal("refresh did not repeat") + } +} +func TestContactDirectoryWorkerClosesAndDoesNotRestart(t *testing.T) { + t.Setenv("OPENMESSAGES_GOOGLE_CONTACTS_MCP_URL", "http://127.0.0.1:1/mcp") + a := newTestApp(t, &mockGMClient{}) + a.StartContactDirectorySync() + a.StartContactDirectorySync() + a.stopContactDirectorySync() + a.StartContactDirectorySync() + if _, err := a.SyncGoogleContacts(); err == nil { + t.Fatal("sync ran after shutdown") + } + if a.GetContactSyncStatus().Running { + t.Fatal("worker left running") + } +} +func TestContactRefreshInterval(t *testing.T) { + for _, s := range []string{"0s", "-1m", "30s", "invalid", "25h"} { + t.Setenv("OPENMESSAGES_CONTACTS_REFRESH_INTERVAL", s) + if contactRefreshInterval() != 5*time.Minute { + t.Fatal(s) + } + } + t.Setenv("OPENMESSAGES_CONTACTS_REFRESH_INTERVAL", "2m") + if contactRefreshInterval() != 2*time.Minute { + t.Fatal("valid interval ignored") + } +} + +func TestContactDirectoryDisabledForNonLegacyDaemon(t *testing.T) { + t.Setenv("OPENMESSAGES_GOOGLE_CONTACTS_MCP_URL", "http://127.0.0.1:1/mcp") + a := newTestApp(t, &mockGMClient{}) + a.ContactDirectoryDisabled = true + a.StartContactDirectorySync() + if a.contactDirectory.cancel != nil { + t.Fatal("started unsupported worker") + } + if _, err := a.SyncGoogleContacts(); err == nil { + t.Fatal("wrote unsupported read store") + } + if a.GetContactSyncStatus().Source != "disabled" { + t.Fatal("misleading status") + } +} diff --git a/internal/app/contacts.go b/internal/app/contacts.go index cc9bde00..5be09b44 100644 --- a/internal/app/contacts.go +++ b/internal/app/contacts.go @@ -1,6 +1,7 @@ package app import ( + "context" "fmt" "strings" @@ -8,6 +9,10 @@ import ( ) func (a *App) StartGoogleContactSync() { + if a != nil && googleContactsMCPURL() != "" { + a.StartContactDirectorySync() + return + } if a == nil || !googleAvatarSyncEnabled() { return } @@ -30,6 +35,9 @@ func (a *App) StartGoogleContactSync() { } func (a *App) SyncGoogleContacts() (int, error) { + if googleContactsMCPURL() != "" { + return a.syncContactDirectory(context.Background()) + } gm := a.getGMClient() if gm == nil { return 0, fmt.Errorf("not connected to Google Messages") diff --git a/internal/contactsync/google.go b/internal/contactsync/google.go new file mode 100644 index 00000000..3dc732e6 --- /dev/null +++ b/internal/contactsync/google.go @@ -0,0 +1,195 @@ +// Package contactsync reads a complete address book from a Google Contacts MCP +// server. Google Messages' LIST_CONTACTS response is only a small suggestion list. +package contactsync + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "os" + "strings" + "time" + + mcpclient "github.com/mark3labs/mcp-go/client" + "github.com/mark3labs/mcp-go/client/transport" + "github.com/mark3labs/mcp-go/mcp" +) + +type Value struct { + Value string `json:"value"` + Type string `json:"type,omitempty"` +} +type Name struct { + DisplayName string `json:"displayName"` +} +type Organization struct { + Name string `json:"name"` + Title string `json:"title,omitempty"` +} +type Person struct { + ResourceName string `json:"resourceName"` + Names []Name `json:"names,omitempty"` + PhoneNumbers []Value `json:"phoneNumbers,omitempty"` + EmailAddresses []Value `json:"emailAddresses,omitempty"` + Organizations []Organization `json:"organizations,omitempty"` +} + +func (p Person) DisplayName() string { + for _, n := range p.Names { + if s := strings.TrimSpace(n.DisplayName); s != "" { + return s + } + } + for _, o := range p.Organizations { + if s := strings.TrimSpace(o.Name); s != "" { + return s + } + } + return "" +} + +type page struct { + Connections []Person `json:"connections"` + NextPageToken string `json:"nextPageToken"` + TotalItems *int `json:"totalItems"` + TotalPeople *int `json:"totalPeople"` +} + +type toolCaller interface { + CallTool(context.Context, mcp.CallToolRequest) (*mcp.CallToolResult, error) +} + +// Fetch follows every page and fails closed on incomplete or inconsistent lists. +// Tokens belong to the connector; OpenMessage never reads Google OAuth cookies. +func Fetch(ctx context.Context, endpoint, tokenFile string) ([]Person, error) { + u, err := url.Parse(endpoint) + if err != nil || u.Host == "" || u.User != nil || u.RawQuery != "" || u.Fragment != "" { + return nil, errors.New("invalid Google Contacts MCP URL") + } + ip := net.ParseIP(u.Hostname()) + if u.Scheme != "https" && !(u.Scheme == "http" && ip != nil && ip.IsLoopback()) { + return nil, errors.New("Google Contacts MCP requires HTTPS or a loopback IP URL") + } + headers := map[string]string{} + if tokenFile != "" { + f, err := os.Open(tokenFile) + if err != nil { + return nil, errors.New("cannot read Google Contacts MCP token file") + } + defer f.Close() + st, err := f.Stat() + if err != nil || !st.Mode().IsRegular() || st.Mode().Perm()&0077 != 0 { + return nil, errors.New("Google Contacts MCP token file must be private (0600)") + } + b, err := io.ReadAll(io.LimitReader(f, 8193)) + if err != nil || len(b) > 8192 { + return nil, errors.New("invalid Google Contacts MCP token file") + } + token := strings.TrimSpace(string(b)) + if token == "" || strings.ContainsAny(token, "\r\n") { + return nil, errors.New("invalid Google Contacts MCP token file") + } + headers["Authorization"] = "Bearer " + token + } + hc := &http.Client{Timeout: 30 * time.Second, Transport: limitedTransport{}, CheckRedirect: func(*http.Request, []*http.Request) error { return errors.New("redirects disabled") }} + cli, err := mcpclient.NewStreamableHttpClient(endpoint, transport.WithHTTPHeaders(headers), transport.WithHTTPBasicClient(hc)) + if err != nil { + return nil, errors.New("cannot configure Google Contacts MCP client") + } + if err = cli.Start(ctx); err != nil { + return nil, errors.New("cannot start Google Contacts MCP client") + } + defer cli.Close() + req := mcp.InitializeRequest{} + req.Params.ProtocolVersion = mcp.LATEST_PROTOCOL_VERSION + req.Params.ClientInfo = mcp.Implementation{Name: "openmessage-contacts", Version: "1"} + if _, err = cli.Initialize(ctx, req); err != nil { + return nil, errors.New("Google Contacts MCP initialization failed; check connector authorization") + } + return fetchPages(ctx, cli) +} + +type limitedTransport struct{} + +func (limitedTransport) RoundTrip(r *http.Request) (*http.Response, error) { + resp, err := http.DefaultTransport.RoundTrip(r) + if err == nil { + resp.Body = &limitedBody{Reader: io.LimitReader(resp.Body, 8<<20), Closer: resp.Body} + } + return resp, err +} + +type limitedBody struct { + io.Reader + io.Closer +} + +func fetchPages(ctx context.Context, cli toolCaller) ([]Person, error) { + people := []Person{} + seenIDs := map[string]bool{} + seenTokens := map[string]bool{} + token := "" + total := -1 + for n := 0; n < 100; n++ { + req := mcp.CallToolRequest{} + req.Params.Name = "contacts_list" + args := map[string]any{"pageSize": 1000, "sortOrder": "LAST_MODIFIED_ASCENDING"} + if token != "" { + args["pageToken"] = token + } + req.Params.Arguments = args + result, err := cli.CallTool(ctx, req) + if err != nil || result == nil || result.IsError { + return nil, errors.New("Google Contacts MCP listing failed; previous directory retained") + } + var raw []byte + if result.StructuredContent != nil { + raw, err = json.Marshal(result.StructuredContent) + } else { + for _, c := range result.Content { + if t, ok := c.(mcp.TextContent); ok { + raw = []byte(t.Text) + break + } + } + } + var p page + if err != nil || len(raw) == 0 || json.Unmarshal(raw, &p) != nil { + return nil, errors.New("invalid Google Contacts MCP listing") + } + t := p.TotalItems + if t == nil { + t = p.TotalPeople + } + if t != nil { + if *t < 0 || (total >= 0 && total != *t) { + return nil, errors.New("Google Contacts listing changed during pagination; retry required") + } + total = *t + } + for _, person := range p.Connections { + if !strings.HasPrefix(person.ResourceName, "people/") || seenIDs[person.ResourceName] { + return nil, errors.New("invalid or repeated Google Contacts resource") + } + seenIDs[person.ResourceName] = true + people = append(people, person) + } + if p.NextPageToken == "" { + if total < 0 || len(people) != total { + return nil, fmt.Errorf("incomplete Google Contacts listing: received %d, expected %d", len(people), total) + } + return people, nil + } + if len(p.Connections) == 0 || seenTokens[p.NextPageToken] { + return nil, errors.New("Google Contacts pagination did not advance") + } + seenTokens[p.NextPageToken] = true + token = p.NextPageToken + } + return nil, errors.New("Google Contacts pagination limit exceeded") +} diff --git a/internal/contactsync/google_test.go b/internal/contactsync/google_test.go new file mode 100644 index 00000000..66e01a23 --- /dev/null +++ b/internal/contactsync/google_test.go @@ -0,0 +1,161 @@ +package contactsync + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/mark3labs/mcp-go/mcp" +) + +type fakeCaller struct { + t *testing.T + pages []*mcp.CallToolResult + tokens []string + calls int + failAt int +} + +func (f *fakeCaller) CallTool(ctx context.Context, r mcp.CallToolRequest) (*mcp.CallToolResult, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + i := f.calls + f.calls++ + if f.failAt > 0 && f.calls == f.failAt { + return nil, errors.New("secret upstream body") + } + if i >= len(f.pages) { + f.t.Fatal("unexpected extra page") + } + a := r.GetArguments() + if a["pageSize"] != 1000 || a["sortOrder"] != "LAST_MODIFIED_ASCENDING" || r.Params.Name != "contacts_list" { + f.t.Fatalf("bad request: %+v", r.Params) + } + if i < len(f.tokens) { + token, _ := a["pageToken"].(string) + if token != f.tokens[i] { + f.t.Fatalf("token %q", token) + } + } + return f.pages[i], nil +} +func pageResult(n, total int, next string, offset int) *mcp.CallToolResult { + p := page{TotalItems: &total, NextPageToken: next, Connections: []Person{}} + for i := 0; i < n; i++ { + p.Connections = append(p.Connections, Person{ResourceName: fmt.Sprintf("people/c%d", offset+i), Names: []Name{{DisplayName: fmt.Sprintf("Person %d", offset+i)}}}) + } + return mcp.NewToolResultStructuredOnly(p) +} +func TestFetchPagesFullDirectory(t *testing.T) { + f := &fakeCaller{t: t, pages: []*mcp.CallToolResult{pageResult(1000, 1002, "next", 0), pageResult(2, 1002, "", 1000)}, tokens: []string{"", "next"}} + p, err := fetchPages(context.Background(), f) + if err != nil || len(p) != 1002 || p[1001].DisplayName() != "Person 1001" { + t.Fatalf("len=%d err=%v", len(p), err) + } +} +func TestFetchPagesRejectsIncomplete(t *testing.T) { + cases := map[string][]*mcp.CallToolResult{ + "missing tail": {pageResult(50, 672, "", 0)}, + "repeated token": {pageResult(1, 3, "next", 0), pageResult(1, 3, "next", 1)}, + "duplicate person": {pageResult(1, 2, "next", 0), pageResult(1, 2, "", 0)}, + "changing total": {pageResult(1, 2, "next", 0), pageResult(1, 3, "", 1)}, + "invalid body": {mcp.NewToolResultText(`{}`)}, + "error body": {mcp.NewToolResultError("private detail must not escape")}, + } + for name, pages := range cases { + t.Run(name, func(t *testing.T) { + p, err := fetchPages(context.Background(), &fakeCaller{t: t, pages: pages}) + if err == nil || p != nil { + t.Fatal("accepted partial list") + } + if strings.Contains(err.Error(), "private detail") { + t.Fatal("leaked error") + } + }) + } + f := &fakeCaller{t: t, pages: []*mcp.CallToolResult{pageResult(1, 2, "next", 0)}, failAt: 2} + p, err := fetchPages(context.Background(), f) + if err == nil || p != nil || strings.Contains(err.Error(), "secret") { + t.Fatal("failed fetch did not fail closed") + } +} +func TestFetchEmptyAndTextResult(t *testing.T) { + for _, r := range []*mcp.CallToolResult{pageResult(0, 0, "", 0), mcp.NewToolResultText(`{"connections":[],"totalPeople":0}`)} { + p, err := fetchPages(context.Background(), &fakeCaller{t: t, pages: []*mcp.CallToolResult{r}}) + if err != nil || len(p) != 0 { + t.Fatalf("%v %v", p, err) + } + } +} +func TestFetchMCPHTTPAndPrivateToken(t *testing.T) { + tokenPath := filepath.Join(t.TempDir(), "token") + if err := os.WriteFile(tokenPath, []byte("test-bearer\n"), 0600); err != nil { + t.Fatal(err) + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer test-bearer" { + t.Error("missing token") + } + if r.Method == http.MethodDelete { + w.WriteHeader(204) + return + } + var req struct { + ID any `json:"id"` + Method string `json:"method"` + } + json.NewDecoder(r.Body).Decode(&req) + w.Header().Set("Content-Type", "application/json") + switch req.Method { + case "initialize": + json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": req.ID, "result": map[string]any{"protocolVersion": "2025-03-26", "capabilities": map[string]any{}, "serverInfo": map[string]any{"name": "test", "version": "1"}}}) + case "notifications/initialized": + w.WriteHeader(202) + case "tools/call": + json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": req.ID, "result": pageResult(2, 2, "", 0)}) + default: + t.Errorf("unexpected method %s", req.Method) + w.WriteHeader(400) + } + })) + defer server.Close() + p, err := Fetch(context.Background(), server.URL, tokenPath) + if err != nil || len(p) != 2 { + t.Fatalf("%d %v", len(p), err) + } + os.Chmod(tokenPath, 0644) + if _, err = Fetch(context.Background(), server.URL, tokenPath); err == nil { + t.Fatal("accepted public token") + } +} +func TestFetchRejectsUnsafeURLs(t *testing.T) { + for _, u := range []string{"http://example.com/mcp", "file:///tmp/a", "https://user:secret@example.com/mcp", "https://example.com/mcp?token=secret"} { + if _, err := Fetch(context.Background(), u, ""); err == nil { + t.Fatal(u) + } + } +} + +func TestFetchDoesNotFollowRedirects(t *testing.T) { + var followed bool + destination := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { followed = true; w.WriteHeader(500) })) + defer destination.Close() + source := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, destination.URL, http.StatusTemporaryRedirect) + })) + defer source.Close() + if _, err := Fetch(context.Background(), source.URL, ""); err == nil { + t.Fatal("redirect accepted") + } + if followed { + t.Fatal("followed connector redirect") + } +} diff --git a/internal/db/contact_directory.go b/internal/db/contact_directory.go new file mode 100644 index 00000000..fff1737d --- /dev/null +++ b/internal/db/contact_directory.go @@ -0,0 +1,210 @@ +package db + +import ( + "encoding/json" + "fmt" + "strings" + "unicode" + + "github.com/maxghenis/openmessage/internal/contactsync" +) + +const directoryPrefix = "google-contacts:" + +// DirectoryPhone rejects extensions, emails and short codes rather than guessing +// an identity. Match full numbers only; retain every original phone in the JSON. +func DirectoryPhone(s string) string { + s = strings.TrimSpace(s) + var b strings.Builder + for _, r := range s { + if r >= '0' && r <= '9' { + b.WriteRune(r) + } else if !strings.ContainsRune("+().- ", r) && !unicode.IsSpace(r) { + return "" + } + } + d := b.String() + if len(d) == 10 && !strings.HasPrefix(s, "+") { + d = "1" + d + } + if len(d) > 15 || (strings.HasPrefix(s, "+") && len(d) < 7) || (!strings.HasPrefix(s, "+") && len(d) < 11) { + return "" + } + return "+" + d +} +func directoryNames(people []contactsync.Person) map[string]string { + names := map[string]string{} + owners := map[string]string{} + for _, p := range people { + for _, n := range p.PhoneNumbers { + phone := DirectoryPhone(n.Value) + if phone == "" { + continue + } + if owner, ok := owners[phone]; ok && owner != p.ResourceName { + names[phone] = "" + continue + } + owners[phone] = p.ResourceName + names[phone] = p.DisplayName() + } + } + return names +} +func (s *Store) loadContactDirectory() error { + rows, err := s.db.Query("SELECT data_json FROM google_contact_directory") + if err != nil { + return err + } + defer rows.Close() + var people []contactsync.Person + for rows.Next() { + var raw string + var p contactsync.Person + if err = rows.Scan(&raw); err != nil { + return err + } + if err = json.Unmarshal([]byte(raw), &p); err != nil { + return err + } + people = append(people, p) + } + if err = rows.Err(); err != nil { + return err + } + s.directoryNames = directoryNames(people) + return nil +} + +// ReplaceContactDirectory atomically replaces only this provider's rows after a +// complete fetch. Existing thread identity, history, titles and read state survive. +func (s *Store) ReplaceContactDirectory(people []contactsync.Person) (int, int, error) { + s.directoryMu.Lock() + defer s.directoryMu.Unlock() + next := directoryNames(people) + tx, err := s.db.Begin() + if err != nil { + return 0, 0, err + } + defer tx.Rollback() + if _, err = tx.Exec("DELETE FROM google_contact_directory"); err != nil { + return 0, 0, err + } + if _, err = tx.Exec("DELETE FROM contacts WHERE substr(contact_id,1,?)=?", len(directoryPrefix), directoryPrefix); err != nil { + return 0, 0, err + } + phones := 0 + for _, p := range people { + raw, e := json.Marshal(p) + if e != nil { + return 0, 0, e + } + if _, err = tx.Exec("INSERT INTO google_contact_directory(resource_name,data_json) VALUES(?,?)", p.ResourceName, string(raw)); err != nil { + return 0, 0, err + } + seen := map[string]bool{} + for _, n := range p.PhoneNumbers { + phone := DirectoryPhone(n.Value) + if phone == "" || seen[phone] { + continue + } + seen[phone] = true + name := p.DisplayName() + if name == "" { + name = phone + } + if _, err = tx.Exec("INSERT INTO contacts(contact_id,name,number) VALUES(?,?,?)", directoryPrefix+p.ResourceName+":"+phone, name, phone); err != nil { + return 0, 0, err + } + phones++ + } + } + rows, err := tx.Query("SELECT conversation_id,name,is_group,participants,source_platform FROM conversations WHERE source_platform IN ('sms','')") + if err != nil { + return 0, 0, err + } + var changed []*Conversation + for rows.Next() { + c := &Conversation{} + if err = rows.Scan(&c.ConversationID, &c.Name, &c.IsGroup, &c.Participants, &c.SourcePlatform); err != nil { + rows.Close() + return 0, 0, err + } + if resolveDirectoryNames(c, s.directoryNames, next) { + changed = append(changed, c) + } + } + err = rows.Err() + rows.Close() + if err != nil { + return 0, 0, err + } + for _, c := range changed { + if _, err = tx.Exec("UPDATE conversations SET name=?,participants=? WHERE conversation_id=?", c.Name, c.Participants, c.ConversationID); err != nil { + return 0, 0, err + } + } + if err = tx.Commit(); err != nil { + return 0, 0, fmt.Errorf("commit contact directory: %w", err) + } + s.directoryNames = next + return phones, len(changed), nil +} + +func replaceDirectoryName(name, phone string, old, next map[string]string) string { + current := next[phone] + if current != "" && (strings.TrimSpace(name) == "" || DirectoryPhone(name) == phone || old[phone] != "" && name == old[phone]) { + return current + } + // Remove a name supplied by the previous directory if the number was deleted + // or became shared/ambiguous. Preserve labels that did not come from it. + if current == "" && old[phone] != "" && name == old[phone] { + return phone + } + return name +} +func resolveDirectoryNames(c *Conversation, old, next map[string]string) bool { + if c.SourcePlatform != "" && c.SourcePlatform != "sms" { + return false + } + var ps []map[string]json.RawMessage + if json.Unmarshal([]byte(c.Participants), &ps) != nil { + return false + } + changed := false + peers := map[string]bool{} + for _, p := range ps { + var me bool + _ = json.Unmarshal(p["is_me"], &me) + if me { + continue + } + var number, name string + _ = json.Unmarshal(p["number"], &number) + _ = json.Unmarshal(p["name"], &name) + phone := DirectoryPhone(number) + if phone == "" { + continue + } + peers[phone] = true + newName := replaceDirectoryName(name, phone, old, next) + if newName != name { + p["name"], _ = json.Marshal(newName) + changed = true + } + } + if !c.IsGroup && len(peers) == 1 { + for phone := range peers { + name := replaceDirectoryName(c.Name, phone, old, next) + if name != c.Name { + c.Name = name + changed = true + } + } + } + if changed { + raw, _ := json.Marshal(ps) + c.Participants = string(raw) + } + return changed +} diff --git a/internal/db/contact_directory_test.go b/internal/db/contact_directory_test.go new file mode 100644 index 00000000..40f3f2ff --- /dev/null +++ b/internal/db/contact_directory_test.go @@ -0,0 +1,140 @@ +package db + +import ( + "encoding/json" + "path/filepath" + "testing" + + "github.com/maxghenis/openmessage/internal/contactsync" +) + +func directoryPerson(id, name string, phones ...string) contactsync.Person { + p := contactsync.Person{ResourceName: "people/" + id, Names: []contactsync.Name{{DisplayName: name}}, Organizations: []contactsync.Organization{{Name: "Example Company", Title: "Engineer"}}, EmailAddresses: []contactsync.Value{{Value: "test@example.com", Type: "work"}}} + for _, n := range phones { + p.PhoneNumbers = append(p.PhoneNumbers, contactsync.Value{Value: n, Type: "mobile"}) + } + return p +} +func seedDirectoryConversation(t *testing.T, s *Store, id, name, phone string, group bool) { + t.Helper() + ps, _ := json.Marshal([]map[string]any{{"name": "Me", "number": "+12025550000", "is_me": true}, {"name": phone, "number": phone, "id": "remote-peer", "contact_id": "phone-local-id", "extra": true}}) + if err := s.UpsertConversation(&Conversation{ConversationID: id, Name: name, Participants: string(ps), IsGroup: group, LastMessageTS: 1234, UnreadCount: 2, IsFavorite: true, NotificationMode: "muted"}); err != nil { + t.Fatal(err) + } +} +func TestContactDirectoryReconcilesAndRetainsMetadata(t *testing.T) { + s := newTestStore(t) + phone := "+12025550101" + seedDirectoryConversation(t, s, "raw", phone, phone, false) + seedDirectoryConversation(t, s, "custom", "My dentist", phone, false) + seedDirectoryConversation(t, s, "group", "Team lunch", phone, true) + if err := s.UpsertContact(&Contact{ContactID: "phone-original", Name: "Original", Number: "+12025550999"}); err != nil { + t.Fatal(err) + } + p := directoryPerson("a", "Alice Example", phone, "+12025550102") + phones, changed, err := s.ReplaceContactDirectory([]contactsync.Person{p}) + if err != nil || phones != 2 || changed != 3 { + t.Fatalf("%d %d %v", phones, changed, err) + } + for id, want := range map[string]string{"raw": "Alice Example", "custom": "My dentist", "group": "Team lunch"} { + c, _ := s.GetConversation(id) + if c.Name != want || c.LastMessageTS != 1234 || c.UnreadCount != 2 || !c.IsFavorite || c.NotificationMode != "muted" { + t.Fatalf("%+v", c) + } + var ps []map[string]any + json.Unmarshal([]byte(c.Participants), &ps) + if ps[1]["extra"] != true || ps[1]["contact_id"] != "phone-local-id" || ps[1]["name"] != "Alice Example" { + t.Fatalf("%v", ps) + } + } + var raw string + if err = s.db.QueryRow("SELECT data_json FROM google_contact_directory").Scan(&raw); err != nil { + t.Fatal(err) + } + var saved contactsync.Person + json.Unmarshal([]byte(raw), &saved) + if len(saved.PhoneNumbers) != 2 || saved.Organizations[0].Name != "Example Company" || saved.EmailAddresses[0].Value != "test@example.com" { + t.Fatalf("%+v", saved) + } + contacts, _ := s.ListContacts("Alice", 10) + if len(contacts) != 2 { + t.Fatal(contacts) + } + // A later phone snapshot still containing a raw number cannot undo resolution. + seedDirectoryConversation(t, s, "raw", phone, phone, false) + c, _ := s.GetConversation("raw") + if c.Name != "Alice Example" { + t.Fatal(c.Name) + } + p.Names[0].DisplayName = "Alice Renamed" + if _, _, err = s.ReplaceContactDirectory([]contactsync.Person{p}); err != nil { + t.Fatal(err) + } + c, _ = s.GetConversation("raw") + if c.Name != "Alice Renamed" { + t.Fatal(c.Name) + } + if _, _, err = s.ReplaceContactDirectory(nil); err != nil { + t.Fatal(err) + } + c, _ = s.GetConversation("raw") + if c.Name != phone { + t.Fatal("deleted identity retained", c.Name) + } + contacts, _ = s.ListContacts("", 10) + if len(contacts) != 1 || contacts[0].ContactID != "phone-original" { + t.Fatal(contacts) + } +} +func TestContactDirectoryAmbiguityRollbackAndRestart(t *testing.T) { + path := filepath.Join(t.TempDir(), "db.sqlite") + s, err := New(path) + if err != nil { + t.Fatal(err) + } + defer func() { s.Close() }() + phone := "+12025550101" + seedDirectoryConversation(t, s, "raw", phone, phone, false) + p := directoryPerson("a", "Alice", phone) + if _, _, err = s.ReplaceContactDirectory([]contactsync.Person{p, p}); err == nil { + t.Fatal("duplicate IDs should rollback") + } + contacts, _ := s.ListContacts("Alice", 10) + if len(contacts) != 0 { + t.Fatal("partial write") + } + if _, _, err = s.ReplaceContactDirectory([]contactsync.Person{p, directoryPerson("b", "Bob", phone)}); err != nil { + t.Fatal(err) + } + c, _ := s.GetConversation("raw") + if c.Name != phone { + t.Fatal("assigned ambiguous identity") + } + if _, _, err = s.ReplaceContactDirectory([]contactsync.Person{p}); err != nil { + t.Fatal(err) + } + s.Close() + s, err = New(path) + if err != nil { + t.Fatal(err) + } + seedDirectoryConversation(t, s, "new", phone, phone, false) + c, _ = s.GetConversation("new") + if c.Name != "Alice" { + t.Fatal("directory not loaded on restart") + } + if _, _, err = s.ReplaceContactDirectory([]contactsync.Person{p, directoryPerson("b", "Bob", phone)}); err != nil { + t.Fatal(err) + } + c, _ = s.GetConversation("raw") + if c.Name != phone { + t.Fatal("new ambiguity retained old name") + } +} +func TestDirectoryPhone(t *testing.T) { + for in, want := range map[string]string{"(202) 555-0101": "+12025550101", "+44 20 7946 0000": "+442079460000", "+49 30 123456": "+4930123456", "2025550101 ext 2": "", "user@example.com": "", "12345": ""} { + if got := DirectoryPhone(in); got != want { + t.Fatalf("%q => %q, want %q", in, got, want) + } + } +} diff --git a/internal/db/conversations.go b/internal/db/conversations.go index 08f6c3ba..ca270cd3 100644 --- a/internal/db/conversations.go +++ b/internal/db/conversations.go @@ -53,6 +53,9 @@ func parseNotificationMode(mode string) (string, error) { } func (s *Store) UpsertConversation(c *Conversation) error { + s.directoryMu.RLock() + defer s.directoryMu.RUnlock() + resolveDirectoryNames(c, nil, s.directoryNames) if c.SourcePlatform == "" { c.SourcePlatform = "sms" } diff --git a/internal/db/db.go b/internal/db/db.go index 20a48486..75bf53d6 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -4,13 +4,16 @@ import ( "database/sql" "fmt" "strings" + "sync" _ "modernc.org/sqlite" ) type Store struct { - db *sql.DB - ftsEnabled bool + db *sql.DB + ftsEnabled bool + directoryMu sync.RWMutex + directoryNames map[string]string } type Conversation struct { @@ -130,6 +133,10 @@ func New(dsn string) (*Store, error) { db.Close() return nil, fmt.Errorf("migrate: %w", err) } + if err := s.loadContactDirectory(); err != nil { + db.Close() + return nil, fmt.Errorf("load contact directory: %w", err) + } return s, nil } @@ -327,6 +334,11 @@ func (s *Store) migrate() error { number TEXT NOT NULL DEFAULT '' ); + CREATE TABLE IF NOT EXISTS google_contact_directory ( + resource_name TEXT PRIMARY KEY, + data_json TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS contact_avatars ( avatar_id TEXT PRIMARY KEY, source_platform TEXT NOT NULL, diff --git a/internal/tools/get_status.go b/internal/tools/get_status.go index 81933de4..5126118e 100644 --- a/internal/tools/get_status.go +++ b/internal/tools/get_status.go @@ -128,6 +128,7 @@ func getStatusHandler(a *app.App, configured ...Options) server.ToolHandlerFunc fmt.Fprintf(&sb, "Data dir: %s\n", a.DataDir) payload := map[string]any{ "overall_connected": overallConnected, + "contact_sync": a.GetContactSyncStatus(), "google": google, "whatsapp": whatsApp, "signal": signal, diff --git a/internal/web/api.go b/internal/web/api.go index e76d3a7c..54084866 100644 --- a/internal/web/api.go +++ b/internal/web/api.go @@ -129,6 +129,7 @@ type APIOptions struct { BackfillStatus func() any // returns a JSON-serializable backfill progress snapshot BackfillPhone func(string) error // targeted backfill for a single phone number SyncGoogleContacts func() (int, error) + ContactSyncStatus func() any } type SearchResult struct { @@ -310,6 +311,9 @@ func APIHandlerWithOptions(store *db.Store, cli *client.Client, logger zerolog.L if opts.GoogleStatus != nil { payload["google"] = opts.GoogleStatus() } + if opts.ContactSyncStatus != nil { + payload["contact_sync"] = opts.ContactSyncStatus() + } if opts.WhatsAppStatus != nil { payload["whatsapp"] = opts.WhatsAppStatus() } diff --git a/internal/web/contact_directory_test.go b/internal/web/contact_directory_test.go new file mode 100644 index 00000000..3b8d19c1 --- /dev/null +++ b/internal/web/contact_directory_test.go @@ -0,0 +1,64 @@ +package web + +import ( + "encoding/json" + "net/http" + "testing" + + "github.com/maxghenis/openmessage/internal/contactsync" + "github.com/maxghenis/openmessage/internal/db" +) + +func TestContactDirectoryRefreshReachesAutocompleteAndExistingThreads(t *testing.T) { + var ts *testServer + state := map[string]any{"complete": false} + ts = newTestServerWithOptions(t, APIOptions{ + ContactSyncStatus: func() any { return state }, + SyncGoogleContacts: func() (int, error) { + n, _, err := ts.store.ReplaceContactDirectory([]contactsync.Person{ + {ResourceName: "people/alice", Names: []contactsync.Name{{DisplayName: "Alice Directory"}}, PhoneNumbers: []contactsync.Value{{Value: "+12025550101"}}}, + {ResourceName: "people/new", Names: []contactsync.Name{{DisplayName: "New Person"}}, PhoneNumbers: []contactsync.Value{{Value: "+12025550102"}, {Value: "+12025550103"}}}, + }) + state = map[string]any{"complete": err == nil, "people": 2, "phone_entries": n} + return n, err + }, + }) + if err := ts.store.UpsertConversation(&db.Conversation{ConversationID: "existing", Name: "+12025550101", Participants: `[{"number":"+12025550101","name":"+12025550101"}]`, LastMessageTS: 1000}); err != nil { + t.Fatal(err) + } + resp, err := http.Post(ts.server.URL+"/api/contacts/sync", "application/json", nil) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != 200 { + t.Fatal(resp.StatusCode) + } + for query, want := range map[string]int{"Alice": 1, "New": 2} { + resp, err = http.Get(ts.server.URL + "/api/contacts?q=" + query) + if err != nil { + t.Fatal(err) + } + var cs []db.Contact + json.NewDecoder(resp.Body).Decode(&cs) + resp.Body.Close() + if len(cs) != want { + t.Fatalf("%s: %+v", query, cs) + } + } + cs, err := ts.store.ListConversations(10) + if err != nil || len(cs) != 1 || cs[0].Name != "Alice Directory" { + t.Fatalf("%+v %v", cs, err) + } + resp, err = http.Get(ts.server.URL + "/api/status") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + var status map[string]json.RawMessage + json.NewDecoder(resp.Body).Decode(&status) + var sync map[string]any + if json.Unmarshal(status["contact_sync"], &sync) != nil || sync["complete"] != true { + t.Fatalf("%s", status["contact_sync"]) + } +}