diff --git a/docs/integration-policy.md b/docs/integration-policy.md index b0cf3698..fcf12301 100644 --- a/docs/integration-policy.md +++ b/docs/integration-policy.md @@ -35,6 +35,12 @@ QR or device-linking flows are acceptable when they are the platform's normal lo Shared or managed app credentials must be injected at release time or supplied through local environment/config overrides for source builds. They must not be committed. +## X + +X archive support ports only the useful normalization behavior from [Birdclaw](https://github.com/steipete/birdclaw) and the small Go-adapter pattern from [Clawdex](https://github.com/openclaw/clawdex). The bundled helper reads an X archive zip directly; Cued does not install, invoke, or read the database of either project. + +Put the exported `twitter*.zip` or `x*archive*.zip` in `~/Downloads`, or set `CUED_X_ARCHIVE_PATH`. Cued imports DM contacts, conversations, messages, and follower/following relationships into `~/.cued/local.db`; current mutual state and its archive generation live in `contact_sources.metadata_json`. Sync is manual because the archive is a point-in-time export: run `cued sync run x` after replacing it. + ## Telegram Telegram is deferred for the public OSS launch. diff --git a/native/helpers/birdclaw-go/.gitignore b/native/helpers/birdclaw-go/.gitignore new file mode 100644 index 00000000..30bcfa4e --- /dev/null +++ b/native/helpers/birdclaw-go/.gitignore @@ -0,0 +1 @@ +.build/ diff --git a/native/helpers/birdclaw-go/LICENSE.openclaw b/native/helpers/birdclaw-go/LICENSE.openclaw new file mode 100644 index 00000000..bde8f2fb --- /dev/null +++ b/native/helpers/birdclaw-go/LICENSE.openclaw @@ -0,0 +1,45 @@ +MIT License + +Copyright (c) 2026 openclaw + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. + +--- + +MIT License + +Copyright (c) 2026 Peter Steinberger + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/native/helpers/birdclaw-go/go.mod b/native/helpers/birdclaw-go/go.mod new file mode 100644 index 00000000..2690d6d2 --- /dev/null +++ b/native/helpers/birdclaw-go/go.mod @@ -0,0 +1,3 @@ +module cued-birdclaw-helper + +go 1.26.0 diff --git a/native/helpers/birdclaw-go/main.go b/native/helpers/birdclaw-go/main.go new file mode 100644 index 00000000..a5971db3 --- /dev/null +++ b/native/helpers/birdclaw-go/main.go @@ -0,0 +1,767 @@ +package main + +import ( + "archive/zip" + "bufio" + "container/heap" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "os" + "path/filepath" + "regexp" + "sort" + "strconv" + "strings" + "time" +) + +const ( + helperVersion = "0.2.0" + protocolVersion = 1 + defaultLimit = 5000 +) + +var ( + dmEntryPattern = regexp.MustCompile(`(?i)(?:^|/)data/direct-messages(?:-group)?(?:-part\d+)?\.js$`) + tweetEntryPattern = regexp.MustCompile(`(?i)(?:^|/)data/tweets(?:-part\d+)?\.js$`) + followerEntryPattern = regexp.MustCompile(`(?i)(?:^|/)data/follower(?:-part\d+)?\.js$`) + followingEntryPattern = regexp.MustCompile(`(?i)(?:^|/)data/following(?:-part\d+)?\.js$`) +) + +type accountRow struct { + ID string `json:"id"` + Name string `json:"name"` + Handle string `json:"handle"` + ExternalUserID string `json:"external_user_id"` +} + +type profileRow struct { + ID string `json:"id"` + Handle string `json:"handle"` + DisplayName string `json:"display_name"` + Bio string `json:"bio"` + FollowersCount int `json:"followers_count"` + FollowingCount int `json:"following_count"` + AvatarURL string `json:"avatar_url"` + Location string `json:"location"` + URL string `json:"url"` + VerifiedType string `json:"verified_type"` +} + +type conversationRow struct { + ID string `json:"id"` + ParticipantIDs []string `json:"participant_ids"` + Title string `json:"title"` + LastMessageAt string `json:"last_message_at"` + IsGroup bool `json:"is_group"` +} + +type messageRow struct { + ID string `json:"id"` + ConversationID string `json:"conversation_id"` + SenderProfileID string `json:"sender_profile_id"` + Text string `json:"text"` + CreatedAt string `json:"created_at"` + Direction string `json:"direction"` +} + +type followEdgeRow struct { + Direction string `json:"direction"` + ProfileID string `json:"profile_id"` + ExternalUserID string `json:"external_user_id"` + Source string `json:"source"` + FirstSeenAt string `json:"first_seen_at"` + LastSeenAt string `json:"last_seen_at"` +} + +type syncResult struct { + HelperVersion string `json:"helperVersion"` + ProtocolVersion int `json:"protocolVersion"` + ArchivePath string `json:"archivePath"` + Generation string `json:"generation"` + NextPhase string `json:"nextPhase,omitempty"` + NextRelationshipUserID string `json:"nextRelationshipUserId,omitempty"` + NextMessageOffset int `json:"nextMessageOffset,omitempty"` + TotalMessageCount int `json:"totalMessageCount"` + TotalRelationshipCount int `json:"totalRelationshipCount"` + Account accountRow `json:"account"` + Profiles []profileRow `json:"profiles"` + Conversations []conversationRow `json:"conversations"` + Messages []messageRow `json:"messages"` + FollowEdges []followEdgeRow `json:"followEdges"` + HasMore bool `json:"hasMore"` +} + +type archiveData struct { + account accountRow + profiles map[string]profileRow + conversations map[string]conversationRow + messages []messageRow + followEdges []followEdgeRow + totalMessages int + totalRelationships int + relationshipHasMore bool +} + +func main() { + if len(os.Args) < 2 { + fail(errors.New("usage: cued-birdclaw-helper ")) + } + switch os.Args[1] { + case "version": + writeJSON(map[string]any{"version": helperVersion, "protocolVersion": protocolVersion}) + case "status": + runStatus(os.Args[2:]) + case "sync": + runSync(os.Args[2:]) + default: + fail(fmt.Errorf("unknown command %q", os.Args[1])) + } +} + +func runStatus(args []string) { + flags := flag.NewFlagSet("status", flag.ContinueOnError) + archivePath := flags.String("archive", "", "X archive zip path") + if err := flags.Parse(args); err != nil { + fail(err) + } + resolved, err := resolveArchivePath(*archivePath) + if err != nil { + fail(err) + } + reader, err := zip.OpenReader(resolved) + if err != nil { + fail(fmt.Errorf("open X archive: %w", err)) + } + defer reader.Close() + if findEntry(reader.File, `data/account.js`) == nil { + fail(errors.New("X archive is missing data/account.js")) + } + writeJSON(map[string]any{ + "helperVersion": helperVersion, "protocolVersion": protocolVersion, + "archivePath": resolved, "available": true, + }) +} + +func runSync(args []string) { + flags := flag.NewFlagSet("sync", flag.ContinueOnError) + archivePath := flags.String("archive", "", "X archive zip path") + messageOffset := flags.Int("message-offset", 0, "message cursor offset") + expectedGeneration := flags.String("generation", "", "archive generation for the cursor") + phase := flags.String("phase", "messages", "messages or relationships") + afterRelationshipUserID := flags.String("after-relationship-user-id", "", "exclusive relationship user id cursor") + knownMessageCount := flags.Int("known-message-count", -1, "known generation message total") + knownRelationshipCount := flags.Int("known-relationship-count", -1, "known generation relationship total") + limit := flags.Int("limit", defaultLimit, "maximum messages per page") + if err := flags.Parse(args); err != nil { + fail(err) + } + if *limit < 1 || *limit > 25000 { + fail(errors.New("limit must be between 1 and 25000")) + } + resolved, err := resolveArchivePath(*archivePath) + if err != nil { + fail(err) + } + generation, err := archiveGeneration(resolved) + if err != nil { + fail(err) + } + if *expectedGeneration != "" && *expectedGeneration != generation { + *messageOffset, *phase, *afterRelationshipUserID = 0, "messages", "" + *knownMessageCount, *knownRelationshipCount = -1, -1 + } + data, err := readArchivePage(resolved, *phase, *messageOffset, *afterRelationshipUserID, *knownMessageCount, *knownRelationshipCount, *limit) + if err != nil { + fail(err) + } + result := syncResult{ + HelperVersion: helperVersion, ProtocolVersion: protocolVersion, ArchivePath: resolved, + Generation: generation, Account: data.account, Profiles: []profileRow{}, + Conversations: []conversationRow{}, Messages: []messageRow{}, FollowEdges: []followEdgeRow{}, + TotalMessageCount: data.totalMessages, TotalRelationshipCount: data.totalRelationships, + } + if *phase == "relationships" { + result.FollowEdges = data.followEdges + result.Profiles = profilesForEdges(result.FollowEdges, data.profiles) + result.HasMore = data.relationshipHasMore + if result.HasMore { + result.NextPhase = "relationships" + if len(data.followEdges) > 0 { + result.NextRelationshipUserID = data.followEdges[len(data.followEdges)-1].ExternalUserID + } + } + } else { + filtered := data.messages + moreMessages := *messageOffset+len(filtered) < data.totalMessages + result.Messages = filtered + result.Conversations, result.Profiles = entitiesForMessages(filtered, data) + result.NextMessageOffset = *messageOffset + len(filtered) + result.HasMore = moreMessages || data.totalRelationships > 0 + if moreMessages { + result.NextPhase = "messages" + } else if data.totalRelationships > 0 { + result.NextPhase = "relationships" + } + } + writeJSON(result) +} + +func entitiesForMessages(messages []messageRow, data archiveData) ([]conversationRow, []profileRow) { + conversationIDs := make(map[string]bool) + profileIDs := make(map[string]bool) + for _, message := range messages { + conversationIDs[message.ConversationID] = true + if message.Direction != "outbound" { + profileIDs[message.SenderProfileID] = true + } + } + conversations := make([]conversationRow, 0, len(conversationIDs)) + for id := range conversationIDs { + conversation := data.conversations[id] + conversations = append(conversations, conversation) + for _, participantID := range conversation.ParticipantIDs { + profileIDs[profileID(participantID)] = true + } + } + profiles := make([]profileRow, 0, len(profileIDs)) + for id := range profileIDs { + if profile, exists := data.profiles[id]; exists { + profiles = append(profiles, profile) + } + } + sort.Slice(conversations, func(i, j int) bool { return conversations[i].ID < conversations[j].ID }) + sort.Slice(profiles, func(i, j int) bool { return profiles[i].ID < profiles[j].ID }) + return conversations, profiles +} + +func profilesForEdges(edges []followEdgeRow, profiles map[string]profileRow) []profileRow { + seen := make(map[string]bool) + result := make([]profileRow, 0, len(edges)) + for _, edge := range edges { + if seen[edge.ProfileID] { + continue + } + seen[edge.ProfileID] = true + result = append(result, profiles[edge.ProfileID]) + } + sort.Slice(result, func(i, j int) bool { return result[i].ID < result[j].ID }) + return result +} + +func archiveGeneration(path string) (string, error) { + info, err := os.Stat(path) + if err != nil { + return "", err + } + return fmt.Sprintf("%d:%d", info.Size(), info.ModTime().UnixNano()), nil +} + +func readArchive(path string) (archiveData, error) { + return readArchivePage(path, "messages", 0, "", -1, -1, int(^uint(0)>>1)) +} + +func readArchivePage(path, phase string, messageOffset int, afterRelationshipUserID string, knownMessageCount, knownRelationshipCount, limit int) (archiveData, error) { + reader, err := zip.OpenReader(path) + if err != nil { + return archiveData{}, fmt.Errorf("open X archive: %w", err) + } + defer reader.Close() + result := archiveData{ + profiles: make(map[string]profileRow), conversations: make(map[string]conversationRow), + } + accountEntry := findEntry(reader.File, `data/account.js`) + if accountEntry == nil { + return archiveData{}, errors.New("X archive is missing data/account.js") + } + if err := forEachRecord(accountEntry, func(record map[string]any) error { + account := asMap(record["account"]) + result.account = accountRow{ + ID: stringValue(account["accountId"]), ExternalUserID: stringValue(account["accountId"]), + Handle: stringValue(account["username"]), + Name: firstNonEmpty(stringValue(account["accountDisplayName"]), stringValue(account["name"]), stringValue(account["username"])), + } + return nil + }); err != nil { + return archiveData{}, err + } + if result.account.ID == "" { + return archiveData{}, errors.New("X archive account id is missing") + } + if phase == "relationships" { + if knownMessageCount >= 0 { + result.totalMessages = knownMessageCount + } + if err := readRelationshipPage(reader.File, &result, afterRelationshipUserID, limit); err != nil { + return archiveData{}, err + } + } else { + for _, file := range reader.File { + if dmEntryPattern.MatchString(normalizePath(file.Name)) { + isGroupSlice := strings.Contains(strings.ToLower(normalizePath(file.Name)), "direct-messages-group") + if err := parseDMEntry(file, &result, messageOffset, limit, isGroupSlice); err != nil { + return archiveData{}, err + } + } + } + if knownRelationshipCount >= 0 { + result.totalRelationships = knownRelationshipCount + } else { + count, err := countRelationshipEdges(reader.File) + if err != nil { + return archiveData{}, err + } + result.totalRelationships = count + } + } + enrichProfilesFromMentions(reader.File, result.profiles) + for id, conversation := range result.conversations { + if conversation.Title == "" { + conversation.Title = conversationTitle(conversation.ParticipantIDs, result.profiles) + result.conversations[id] = conversation + } + } + return result, nil +} + +func parseDMEntry(file *zip.File, result *archiveData, messageOffset, messageLimit int, isGroupSlice bool) error { + reader, decoder, err := archiveArrayDecoder(file) + if err != nil { + return err + } + defer reader.Close() + for decoder.More() { + if token, err := decoder.Token(); err != nil || token != json.Delim('{') { + return fmt.Errorf("read %s record: %w", file.Name, err) + } + for decoder.More() { + keyToken, err := decoder.Token() + if err != nil { + return err + } + if keyToken != "dmConversation" { + var ignored any + if err := decoder.Decode(&ignored); err != nil { + return err + } + continue + } + if err := parseDMConversation(decoder, result, messageOffset, messageLimit, isGroupSlice); err != nil { + return err + } + } + if _, err := decoder.Token(); err != nil { + return err + } + } + return nil +} + +func parseDMConversation(decoder *json.Decoder, result *archiveData, messageOffset, messageLimit int, isGroupSlice bool) error { + if token, err := decoder.Token(); err != nil || token != json.Delim('{') { + return fmt.Errorf("read dmConversation: %w", err) + } + conversationID, name := "", "" + membership := map[string]bool{} + knownMembership := map[string]bool{} + selected := []messageRow{} + lastMessageAt := "" + for decoder.More() { + keyToken, err := decoder.Token() + if err != nil { + return err + } + key := keyToken.(string) + switch key { + case "conversationId": + if err := decoder.Decode(&conversationID); err != nil { + return err + } + case "name": + if err := decoder.Decode(&name); err != nil { + return err + } + case "messages": + if token, err := decoder.Token(); err != nil || token != json.Delim('[') { + return err + } + for decoder.More() { + var event map[string]any + if err := decoder.Decode(&event); err != nil { + return err + } + applyMembershipEvent(event, membership, knownMembership) + messageCreate := asMap(event["messageCreate"]) + if len(messageCreate) == 0 { + continue + } + messageIndex := result.totalMessages + result.totalMessages++ + if messageIndex < messageOffset || len(result.messages)+len(selected) >= messageLimit { + continue + } + senderID := stringValue(messageCreate["senderId"]) + createdAt := parseTwitterDate(stringValue(messageCreate["createdAt"])) + direction := "inbound" + if senderID == result.account.ExternalUserID { + direction = "outbound" + } + selected = append(selected, messageRow{ID: firstNonEmpty(stringValue(messageCreate["id"]), "message-"+strconv.Itoa(messageIndex)), SenderProfileID: profileID(senderID), Text: stringValue(messageCreate["text"]), CreatedAt: createdAt, Direction: direction}) + if createdAt > lastMessageAt { + lastMessageAt = createdAt + } + } + if _, err := decoder.Token(); err != nil { + return err + } + default: + var ignored any + if err := decoder.Decode(&ignored); err != nil { + return err + } + } + } + if _, err := decoder.Token(); err != nil { + return err + } + if conversationID == "" || len(selected) == 0 { + return nil + } + participants := []string{} + for userID, active := range membership { + if active && userID != "" && userID != result.account.ExternalUserID { + participants = append(participants, userID) + ensureProfile(result, userID) + } + } + sort.Strings(participants) + for index := range selected { + selected[index].ConversationID = conversationID + if strings.HasPrefix(selected[index].ID, "message-") { + selected[index].ID = conversationID + "-" + selected[index].ID + } + if selected[index].Direction != "outbound" { + ensureProfile(result, strings.TrimPrefix(selected[index].SenderProfileID, "profile_user_")) + } + } + result.messages = append(result.messages, selected...) + result.conversations[conversationID] = conversationRow{ID: conversationID, ParticipantIDs: participants, Title: name, LastMessageAt: lastMessageAt, IsGroup: isGroupSlice || len(participants) > 1 || name != ""} + return nil +} + +func applyMembershipEvent(event map[string]any, membership, known map[string]bool) { + // X stores DM events newest-first, so the first membership fact for a user is current. + setIfUnknown := func(userID string, active bool) { + if userID == "" || known[userID] { + return + } + known[userID], membership[userID] = true, active + } + if payload := asMap(event["participantsLeave"]); len(payload) > 0 { + userIDs := asSlice(payload["userIds"]) + for _, value := range userIDs { + setIfUnknown(stringValue(value), false) + } + if len(userIDs) == 0 { + setIfUnknown(stringValue(payload["initiatingUserId"]), false) + } + } + if payload := asMap(event["participantsJoin"]); len(payload) > 0 { + for _, value := range asSlice(payload["userIds"]) { + setIfUnknown(stringValue(value), true) + } + setIfUnknown(stringValue(payload["initiatingUserId"]), true) + } + if payload := asMap(event["joinConversation"]); len(payload) > 0 { + for _, value := range asSlice(payload["participantsSnapshot"]) { + setIfUnknown(stringValue(value), true) + } + } + if payload := asMap(event["messageCreate"]); len(payload) > 0 { + setIfUnknown(stringValue(payload["senderId"]), true) + setIfUnknown(stringValue(payload["recipientId"]), true) + } +} + +type userIDMaxHeap []string + +func (items userIDMaxHeap) Len() int { return len(items) } +func (items userIDMaxHeap) Less(i, j int) bool { return items[i] > items[j] } +func (items userIDMaxHeap) Swap(i, j int) { items[i], items[j] = items[j], items[i] } +func (items *userIDMaxHeap) Push(value any) { *items = append(*items, value.(string)) } +func (items *userIDMaxHeap) Pop() any { + old := *items + last := old[len(old)-1] + *items = old[:len(old)-1] + return last +} + +var followInputs = []struct { + pattern *regexp.Regexp + direction, key string +}{ + {followerEntryPattern, "followers", "follower"}, + {followingEntryPattern, "following", "following"}, +} + +func scanFollowRecords(files []*zip.File, visit func(direction, userID string)) error { + for _, input := range followInputs { + for _, file := range files { + if !input.pattern.MatchString(normalizePath(file.Name)) { + continue + } + if err := forEachRecord(file, func(record map[string]any) error { + userID := stringValue(asMap(record[input.key])["accountId"]) + if userID != "" { + visit(input.direction, userID) + } + return nil + }); err != nil { + return err + } + } + } + return nil +} + +func countRelationshipEdges(files []*zip.File) (int, error) { + count := 0 + err := scanFollowRecords(files, func(_, _ string) { count++ }) + return count, err +} + +func readRelationshipPage(files []*zip.File, result *archiveData, afterUserID string, limit int) error { + candidates := &userIDMaxHeap{} + heap.Init(candidates) + selected := map[string]bool{} + total := 0 + if err := scanFollowRecords(files, func(_ string, userID string) { + total++ + if userID <= afterUserID || selected[userID] { + return + } + if candidates.Len() < limit { + heap.Push(candidates, userID) + selected[userID] = true + return + } + if userID >= (*candidates)[0] { + return + } + delete(selected, heap.Pop(candidates).(string)) + heap.Push(candidates, userID) + selected[userID] = true + }); err != nil { + return err + } + userIDs := make([]string, 0, len(selected)) + for userID := range selected { + userIDs = append(userIDs, userID) + } + sort.Strings(userIDs) + directions := map[string]map[string]bool{} + if err := scanFollowRecords(files, func(direction, userID string) { + if !selected[userID] { + return + } + if directions[userID] == nil { + directions[userID] = map[string]bool{} + } + directions[userID][direction] = true + }); err != nil { + return err + } + for _, userID := range userIDs { + ensureProfile(result, userID) + for _, direction := range []string{"followers", "following"} { + if directions[userID][direction] { + result.followEdges = append(result.followEdges, followEdgeRow{Direction: direction, ProfileID: profileID(userID), ExternalUserID: userID, Source: "archive"}) + } + } + } + result.totalRelationships = total + if len(userIDs) > 0 { + last := userIDs[len(userIDs)-1] + if err := scanFollowRecords(files, func(_ string, userID string) { + if userID > last { + result.relationshipHasMore = true + } + }); err != nil { + return err + } + } + return nil +} + +func enrichProfilesFromMentions(files []*zip.File, profiles map[string]profileRow) { + for _, file := range files { + if !tweetEntryPattern.MatchString(normalizePath(file.Name)) { + continue + } + _ = forEachRecord(file, func(record map[string]any) error { + tweet := asMap(record["tweet"]) + entities := asMap(tweet["entities"]) + for _, rawMention := range asSlice(entities["user_mentions"]) { + mention := asMap(rawMention) + userID := firstNonEmpty(stringValue(mention["id_str"]), stringValue(mention["id"])) + id := profileID(userID) + if _, wanted := profiles[id]; wanted { + handle := stringValue(mention["screen_name"]) + profiles[id] = profileRow{ + ID: id, Handle: handle, + DisplayName: firstNonEmpty(stringValue(mention["name"]), handle, "X user "+userID), + } + } + } + return nil + }) + } +} + +func ensureProfile(result *archiveData, userID string) { + id := profileID(userID) + if _, exists := result.profiles[id]; exists { + return + } + result.profiles[id] = profileRow{ID: id, DisplayName: "X user " + userID} +} + +func archiveArrayDecoder(file *zip.File) (io.ReadCloser, *json.Decoder, error) { + reader, err := file.Open() + if err != nil { + return nil, nil, err + } + buffered := bufio.NewReader(reader) + if _, err := buffered.ReadString('='); err != nil { + reader.Close() + return nil, nil, fmt.Errorf("read %s assignment: %w", file.Name, err) + } + decoder := json.NewDecoder(buffered) + if token, err := decoder.Token(); err != nil || token != json.Delim('[') { + reader.Close() + return nil, nil, fmt.Errorf("read %s array: %w", file.Name, err) + } + return reader, decoder, nil +} + +func forEachRecord(file *zip.File, visit func(map[string]any) error) error { + reader, err := file.Open() + if err != nil { + return err + } + defer reader.Close() + buffered := bufio.NewReader(reader) + if _, err := buffered.ReadString('='); err != nil { + return fmt.Errorf("read %s assignment: %w", file.Name, err) + } + decoder := json.NewDecoder(buffered) + token, err := decoder.Token() + if err != nil || token != json.Delim('[') { + return fmt.Errorf("read %s array: %w", file.Name, err) + } + for decoder.More() { + var record map[string]any + if err := decoder.Decode(&record); err != nil { + return fmt.Errorf("decode %s: %w", file.Name, err) + } + if err := visit(record); err != nil { + return err + } + } + return nil +} + +func findEntry(files []*zip.File, suffix string) *zip.File { + suffix = strings.ToLower(suffix) + for _, file := range files { + if strings.HasSuffix(strings.ToLower(normalizePath(file.Name)), suffix) { + return file + } + } + return nil +} + +func resolveArchivePath(value string) (string, error) { + path := strings.TrimSpace(value) + if path == "" { + return "", errors.New("X archive path is required") + } + if strings.HasPrefix(path, "~/") { + home, err := os.UserHomeDir() + if err != nil { + return "", err + } + path = filepath.Join(home, strings.TrimPrefix(path, "~/")) + } + return filepath.Abs(path) +} + +func parseTwitterDate(value string) string { + for _, layout := range []string{time.RFC3339Nano, "Mon Jan 02 15:04:05 -0700 2006"} { + if parsed, err := time.Parse(layout, value); err == nil { + return parsed.UTC().Format(time.RFC3339Nano) + } + } + return value +} + +func profileID(userID string) string { return "profile_user_" + userID } +func normalizePath(value string) string { return strings.ReplaceAll(value, `\`, "/") } + +func conversationTitle(participants []string, profiles map[string]profileRow) string { + names := make([]string, 0, len(participants)) + for _, userID := range participants { + names = append(names, profiles[profileID(userID)].DisplayName) + } + return strings.Join(names, ", ") +} + +func asMap(value any) map[string]any { + if result, ok := value.(map[string]any); ok { + return result + } + return map[string]any{} +} + +func asSlice(value any) []any { + if result, ok := value.([]any); ok { + return result + } + return nil +} + +func stringValue(value any) string { + switch typed := value.(type) { + case string: + return typed + case json.Number: + return typed.String() + case float64: + return strconv.FormatInt(int64(typed), 10) + default: + return "" + } +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + if strings.TrimSpace(value) != "" { + return strings.TrimSpace(value) + } + } + return "" +} + +func writeJSON(value any) { + if err := json.NewEncoder(os.Stdout).Encode(value); err != nil { + fail(err) + } +} + +func fail(err error) { + _ = json.NewEncoder(os.Stdout).Encode(map[string]any{"ok": false, "error": err.Error()}) + os.Exit(1) +} diff --git a/native/helpers/birdclaw-go/main_test.go b/native/helpers/birdclaw-go/main_test.go new file mode 100644 index 00000000..23439816 --- /dev/null +++ b/native/helpers/birdclaw-go/main_test.go @@ -0,0 +1,156 @@ +package main + +import ( + "archive/zip" + "os" + "path/filepath" + "testing" +) + +func TestReadsArchiveDMsAndFollowGraph(t *testing.T) { + archivePath := filepath.Join(t.TempDir(), "twitter-archive.zip") + file, err := os.Create(archivePath) + if err != nil { + t.Fatal(err) + } + writer := zip.NewWriter(file) + entries := map[string]string{ + "archive/data/account.js": `window.YTD.account.part0 = [{"account":{"accountId":"10","username":"sam","accountDisplayName":"Sam"}}]`, + "archive/data/tweets.js": `window.YTD.tweets.part0 = [{"tweet":{"entities":{"user_mentions":[{"id_str":"20","screen_name":"ava","name":"Ava Chen"}]}}}]`, + "archive/data/direct-messages-group.js": `window.YTD.direct_messages_group.part0 = [{"dmConversation":{"conversationId":"group-1","messages":[{"participantsLeave":{"initiatingUserId":"30","userIds":["30","40"]}},{"messageCreate":{"id":"m1","senderId":"20","recipientId":"10","text":"hello","createdAt":"2026-08-03T12:00:00.000Z"}},{"participantsJoin":{"initiatingUserId":"20","userIds":["40"]}},{"joinConversation":{"participantsSnapshot":["10","20","30"]}}]}}]`, + "archive/data/follower.js": `window.YTD.follower.part0 = [{"follower":{"accountId":"20"}}]`, + "archive/data/following.js": `window.YTD.following.part0 = [{"following":{"accountId":"20"}}]`, + } + for name, content := range entries { + entry, err := writer.Create(name) + if err != nil { + t.Fatal(err) + } + if _, err := entry.Write([]byte(content)); err != nil { + t.Fatal(err) + } + } + if err := writer.Close(); err != nil { + t.Fatal(err) + } + if err := file.Close(); err != nil { + t.Fatal(err) + } + + data, err := readArchive(archivePath) + if err != nil { + t.Fatal(err) + } + if data.account.ID != "10" || data.account.Handle != "sam" { + t.Fatalf("account = %#v", data.account) + } + if len(data.messages) != 1 || data.messages[0].Text != "hello" { + t.Fatalf("messages = %#v", data.messages) + } + conversation := data.conversations["group-1"] + if len(conversation.ParticipantIDs) != 1 || conversation.ParticipantIDs[0] != "20" { + t.Fatalf("conversation participants = %#v", conversation.ParticipantIDs) + } + if !conversation.IsGroup || conversation.Title != "Ava Chen" { + t.Fatalf("conversation = %#v", conversation) + } + profile := data.profiles["profile_user_20"] + if profile.Handle != "ava" || profile.DisplayName != "Ava Chen" { + t.Fatalf("profile = %#v", profile) + } + if data.totalRelationships != 2 { + t.Fatalf("totalRelationships = %d", data.totalRelationships) + } + relationships, err := readArchivePage(archivePath, "relationships", 0, "", data.totalMessages, data.totalRelationships, 1) + if err != nil { + t.Fatal(err) + } + if len(relationships.followEdges) != 2 { + t.Fatalf("followEdges = %#v", relationships.followEdges) + } +} + +func TestReadsOnlyRequestedMessagePage(t *testing.T) { + archivePath := filepath.Join(t.TempDir(), "twitter-archive.zip") + file, err := os.Create(archivePath) + if err != nil { + t.Fatal(err) + } + writer := zip.NewWriter(file) + entries := map[string]string{ + "data/account.js": `window.YTD.account.part0 = [{"account":{"accountId":"10","username":"sam"}}]`, + "data/direct-messages.js": `window.YTD.direct_messages.part0 = [{"dmConversation":{"conversationId":"10-20","messages":[{"messageCreate":{"id":"m1","senderId":"20","recipientId":"10"}},{"messageCreate":{"id":"m2","senderId":"20","recipientId":"10"}},{"messageCreate":{"id":"m3","senderId":"20","recipientId":"10"}}]}}]`, + } + for name, content := range entries { + entry, createErr := writer.Create(name) + if createErr != nil { + t.Fatal(createErr) + } + if _, writeErr := entry.Write([]byte(content)); writeErr != nil { + t.Fatal(writeErr) + } + } + if err := writer.Close(); err != nil { + t.Fatal(err) + } + if err := file.Close(); err != nil { + t.Fatal(err) + } + data, err := readArchivePage(archivePath, "messages", 1, "", -1, -1, 1) + if err != nil { + t.Fatal(err) + } + if data.totalMessages != 3 || len(data.messages) != 1 || data.messages[0].ID != "m2" { + t.Fatalf("page = total %d messages %#v", data.totalMessages, data.messages) + } +} + +func TestPagesRelationshipsWithoutParsingDMs(t *testing.T) { + archivePath := filepath.Join(t.TempDir(), "twitter-archive.zip") + file, err := os.Create(archivePath) + if err != nil { + t.Fatal(err) + } + writer := zip.NewWriter(file) + entries := map[string]string{ + "data/account.js": `window.YTD.account.part0 = [{"account":{"accountId":"10","username":"sam"}}]`, + "data/direct-messages.js": `this is intentionally not a valid archive slice`, + "data/follower.js": `window.YTD.follower.part0 = [{"follower":{"accountId":"30"}},{"follower":{"accountId":"10"}},{"follower":{"accountId":"20"}},{"follower":{"accountId":"40"}}]`, + "data/following.js": `window.YTD.following.part0 = [{"following":{"accountId":"30"}},{"following":{"accountId":"20"}}]`, + } + for name, content := range entries { + entry, createErr := writer.Create(name) + if createErr != nil { + t.Fatal(createErr) + } + if _, writeErr := entry.Write([]byte(content)); writeErr != nil { + t.Fatal(writeErr) + } + } + if err := writer.Close(); err != nil { + t.Fatal(err) + } + if err := file.Close(); err != nil { + t.Fatal(err) + } + first, err := readArchivePage(archivePath, "relationships", 0, "", 7, 6, 2) + if err != nil { + t.Fatal(err) + } + if !first.relationshipHasMore || len(first.profiles) != 2 || len(first.followEdges) != 3 { + t.Fatalf("first relationship page = %#v", first) + } + second, err := readArchivePage(archivePath, "relationships", 0, "20", 7, 6, 2) + if err != nil { + t.Fatal(err) + } + if second.relationshipHasMore || len(second.profiles) != 2 || len(second.followEdges) != 3 { + t.Fatalf("second relationship page = %#v", second) + } +} + +func TestParseTwitterDate(t *testing.T) { + if got := parseTwitterDate("Mon Feb 19 14:59:19 +0000 2024"); got != "2024-02-19T14:59:19Z" { + t.Fatalf("parseTwitterDate = %q", got) + } +} diff --git a/native/macos/CuedNative/Sources/Interface/Components/PlatformIcon.swift b/native/macos/CuedNative/Sources/Interface/Components/PlatformIcon.swift index 310650e0..768799b7 100644 --- a/native/macos/CuedNative/Sources/Interface/Components/PlatformIcon.swift +++ b/native/macos/CuedNative/Sources/Interface/Components/PlatformIcon.swift @@ -103,6 +103,10 @@ public struct PlatformIcon: View { .font(.system(size: 14, weight: .black, design: .rounded)) .foregroundStyle(.white) .offset(y: 0.5) + case "x": + Text("X") + .font(.system(size: 14, weight: .bold, design: .rounded)) + .foregroundStyle(.white) case "signal": ZStack { ZStack { diff --git a/native/macos/CuedNative/Sources/Interface/Root/RootDisplayHelpers.swift b/native/macos/CuedNative/Sources/Interface/Root/RootDisplayHelpers.swift index 17864672..2efb8cf6 100644 --- a/native/macos/CuedNative/Sources/Interface/Root/RootDisplayHelpers.swift +++ b/native/macos/CuedNative/Sources/Interface/Root/RootDisplayHelpers.swift @@ -120,6 +120,8 @@ func platformAccentColor(for platform: String) -> Color { return Color(red: 0.24, green: 0.56, blue: 0.98) case "whatsapp": return Color(red: 0.13, green: 0.74, blue: 0.38) + case "x": + return Color(red: 0.08, green: 0.08, blue: 0.09) default: return .accentColor } @@ -239,6 +241,8 @@ func platformDescription(for configuration: PlatformConfig) -> String { return "Syncs WhatsApp messages after you link your phone." case "signal": return "Syncs Signal messages after you link your device." + case "x": + return "Imports direct messages and follow relationships from your local X archive." default: return "Syncs this source on this Mac." } @@ -331,6 +335,8 @@ func platformTitle(_ platform: String, fallback: String?) -> String { return "Discord" case "whatsapp": return "WhatsApp" + case "x": + return "X" default: if let fallback = fallback?.trimmingCharacters(in: .whitespacesAndNewlines), !fallback.isEmpty { return fallback diff --git a/scripts/build-cued-daemon-app.sh b/scripts/build-cued-daemon-app.sh index ee8e6636..d4871b2c 100644 --- a/scripts/build-cued-daemon-app.sh +++ b/scripts/build-cued-daemon-app.sh @@ -29,6 +29,7 @@ APP_PERMISSIONS_ENTITLEMENTS="$ROOT_DIR/scripts/packaging/app-permissions.entitl SIGNAL_HELPER_SOURCE_DIR="$ROOT_DIR/native/helpers/signal-cli/.build/cued-signal-cli" SLACK_HELPER_SOURCE="$ROOT_DIR/native/helpers/slack-go/.build/cued-slack-helper" WHATSAPP_HELPER_SOURCE="$ROOT_DIR/native/helpers/whatsapp-go/.build/cued-whatsapp-helper" +BIRDCLAW_HELPER_SOURCE="$ROOT_DIR/native/helpers/birdclaw-go/.build/cued-birdclaw-helper" PERMISSIONS_SCRIPT_SOURCE="$ROOT_DIR/scripts/request-macos-access.sh" APP_ICON_SOURCE="$ROOT_DIR/native/macos/CuedNative/Resources/AppIcon.icns" TRAY_ICON_SOURCE="$ROOT_DIR/native/macos/CuedNative/Resources/trayIconTemplate.png" @@ -240,6 +241,8 @@ mkdir -p "$(dirname "$SLACK_HELPER_SOURCE")" (cd "$ROOT_DIR/native/helpers/slack-go" && GOWORK=off go build -o "$SLACK_HELPER_SOURCE" .) >/dev/null mkdir -p "$(dirname "$WHATSAPP_HELPER_SOURCE")" (cd "$ROOT_DIR/native/helpers/whatsapp-go" && GOWORK=off go build -o "$WHATSAPP_HELPER_SOURCE" .) >/dev/null +mkdir -p "$(dirname "$BIRDCLAW_HELPER_SOURCE")" +(cd "$ROOT_DIR/native/helpers/birdclaw-go" && GOWORK=off go build -o "$BIRDCLAW_HELPER_SOURCE" .) >/dev/null npm_config_ignore_scripts=true pnpm --dir "$ROOT_DIR" --filter . deploy --legacy --prod "$DEPLOY_STAGING_DIR" >/dev/null copy_better_sqlite3_binary @@ -270,6 +273,8 @@ cp "$SLACK_HELPER_SOURCE" "$HELPERS_DIR/cued-slack-helper" chmod +x "$HELPERS_DIR/cued-slack-helper" cp "$WHATSAPP_HELPER_SOURCE" "$HELPERS_DIR/cued-whatsapp-helper" chmod +x "$HELPERS_DIR/cued-whatsapp-helper" +cp "$BIRDCLAW_HELPER_SOURCE" "$HELPERS_DIR/cued-birdclaw-helper" +chmod +x "$HELPERS_DIR/cued-birdclaw-helper" if [[ -z "$BETTER_SQLITE3_BINDING_SOURCE" ]]; then echo "better-sqlite3-multiple-ciphers native binding not found in node_modules" >&2 @@ -418,6 +423,7 @@ if [[ -f "\$SCRIPT_DIR/oauth/google-oauth-client.json" ]]; then fi export CUED_SLACK_HELPER_BINARY="\${CUED_SLACK_HELPER_BINARY:-\$SCRIPT_DIR/helpers/cued-slack-helper}" export CUED_WHATSAPP_HELPER_BINARY="\${CUED_WHATSAPP_HELPER_BINARY:-\$SCRIPT_DIR/helpers/cued-whatsapp-helper}" +export CUED_X_ARCHIVE_HELPER_BINARY="\${CUED_X_ARCHIVE_HELPER_BINARY:-\$SCRIPT_DIR/helpers/cued-birdclaw-helper}" export CUED_APP_VERSION="$APP_VERSION" export CUED_RELEASE_CHANNEL="$RELEASE_CHANNEL" exec "\$NODE_BIN" "\$RUNTIME_ROOT/dist/cli.js" "\$@" diff --git a/skills/cued/SKILL.md b/skills/cued/SKILL.md index 0758d445..7fb431e9 100644 --- a/skills/cued/SKILL.md +++ b/skills/cued/SKILL.md @@ -1,6 +1,6 @@ --- name: cued -description: Queries Cued through the local `cued` CLI for the user's real contacts, conversations, and messages synced from iMessage, Slack, WhatsApp, LinkedIn, Gmail, and Signal. ALWAYS use this skill when the user asks anything about their contacts, messages, texts, conversations, or communication - even short queries like "who texted me", "check my messages", "what's John's email", or "what did we talk about on Slack". Covers finding contacts, looking up phone numbers and email addresses, reading message history, follow-up detection, ghosting detection, dormant relationships, network search, unread triage, cross-platform conversation lookup, contact deduplication, attachment lookup/fetch, and relationship analysis. The database has real data - do not tell the user you lack access to their messages or contacts. +description: Queries Cued through the local `cued` CLI for the user's real contacts, conversations, and messages synced from iMessage, Slack, WhatsApp, LinkedIn, Gmail, Signal, and X. ALWAYS use this skill when the user asks anything about their contacts, messages, texts, conversations, or communication - even short queries like "who texted me", "check my messages", "what's John's email", or "what did we talk about on Slack". Covers finding contacts, looking up phone numbers and email addresses, reading message history, follow-up detection, ghosting detection, dormant relationships, network search, unread triage, cross-platform conversation lookup, contact deduplication, attachment lookup/fetch, and relationship analysis. The database has real data - do not tell the user you lack access to their messages or contacts. --- # Cued diff --git a/src/core/types/provider.ts b/src/core/types/provider.ts index 1eb9082b..e481a5db 100644 --- a/src/core/types/provider.ts +++ b/src/core/types/provider.ts @@ -23,6 +23,11 @@ export interface ContactObservationPayload { fields: ContactFields; handles: ContactHandleInput[]; sourceProfileUrl?: string | null; + sourceMetadata?: Record | null; +} + +export interface ContactSnapshotCompletedPayload { + generation: string; } export interface ConversationParticipantInput { @@ -39,6 +44,7 @@ export interface ConversationObservationPayload { topic?: string | null; unreadCount?: number | null; removalReason?: string | null; + participantsAreSnapshot?: boolean; participants: ConversationParticipantInput[]; } @@ -142,6 +148,7 @@ export interface CallPayload { export type RawEventPayload = | ContactObservationPayload + | ContactSnapshotCompletedPayload | ConversationObservationPayload | CallPayload | MessagePayload diff --git a/src/platforms/core/proofs.ts b/src/platforms/core/proofs.ts index 9ace7214..2bcf7751 100644 --- a/src/platforms/core/proofs.ts +++ b/src/platforms/core/proofs.ts @@ -109,6 +109,14 @@ const PROOF_KIND_CONTRACTS: SyncProofKindContract[] = [ invalidatedBy: "A later helper resync reports additional pages or a newer message range.", resumeCursorMeans: "The WhatsApp helper resync cursor and selected since timestamp.", }, + { + platform: "x", + proofKind: "messages", + scopeKind: "account", + completeMeans: "All DMs and relationship edges in the selected X archive generation were read.", + invalidatedBy: "The selected X archive zip is replaced with a different generation.", + resumeCursorMeans: "The archive generation plus DM or relationship page boundary.", + }, ]; export function listSyncProofKindContracts(platform?: AdapterPlatform): SyncProofKindContract[] { diff --git a/src/platforms/core/registry.ts b/src/platforms/core/registry.ts index 81b7a7a5..7b508ace 100644 --- a/src/platforms/core/registry.ts +++ b/src/platforms/core/registry.ts @@ -66,6 +66,12 @@ const ADAPTER_DEFINITIONS: Record = { autoSync: true, workerTimeoutMs: 60_000, }, + x: { + platform: "x", + workerEntrypoint: join(MODULE_DIRNAME, "../x/sync/worker.js"), + autoSync: false, + workerTimeoutMs: 120_000, + }, }; export function listAdapterPlatforms(): AdapterPlatform[] { diff --git a/src/platforms/core/state/integration-state.test.ts b/src/platforms/core/state/integration-state.test.ts index 82c9b1f4..5508e4f0 100644 --- a/src/platforms/core/state/integration-state.test.ts +++ b/src/platforms/core/state/integration-state.test.ts @@ -204,7 +204,7 @@ process.exit(44); const db = createDb(); const refreshed = await refreshManagedIntegrationStates(db); - expect(refreshed.refreshed).toBe(4); + expect(refreshed.refreshed).toBe(5); expect(listIntegrationStates(db)).toEqual( expect.arrayContaining([ expect.objectContaining({ @@ -310,6 +310,7 @@ process.exit(44); "linkedin", "whatsapp", "signal", + "x", ]); db.close(); }); @@ -330,6 +331,7 @@ process.exit(44); "linkedin", "whatsapp", "signal", + "x", ]); db.close(); @@ -736,7 +738,7 @@ process.exit(44); const refreshed = refreshLocalIntegrationStates(db); - expect(refreshed.refreshed).toBe(2); + expect(refreshed.refreshed).toBe(3); expect(listIntegrationStates(db)).toEqual( expect.arrayContaining([ expect.objectContaining({ @@ -1668,7 +1670,7 @@ process.exit(44); const refreshed = await refreshManagedIntegrationStates(db); - expect(refreshed.refreshed).toBe(10); + expect(refreshed.refreshed).toBe(11); expect(listIntegrationStates(db)).toEqual( expect.arrayContaining([ expect.objectContaining({ diff --git a/src/platforms/core/state/local.ts b/src/platforms/core/state/local.ts index bb2500b1..0be35352 100644 --- a/src/platforms/core/state/local.ts +++ b/src/platforms/core/state/local.ts @@ -2,6 +2,7 @@ import { execFileSync } from "node:child_process"; import { existsSync } from "node:fs"; import { resolveMacOSNativeBinary } from "../../../runtime/native-binary.js"; import { DEFAULT_CHAT_DB_PATH, IMessageReader } from "../../imessage/reader.js"; +import { inspectBirdclawHelper } from "../../x/helper/binary.js"; import { type IntegrationAuthState, parseIntegrationAuthState } from "../types.js"; import type { ManagedIntegrationState } from "./types.js"; @@ -82,6 +83,7 @@ export function buildLocalIntegrationStates(): ManagedIntegrationState[] { const chatDbPath = process.env.CUED_IMESSAGE_DB_PATH ?? DEFAULT_CHAT_DB_PATH; const contactsAuthState = getContactsAuthState(); const imessageAuthState = getIMessageAuthState(); + const birdclaw = inspectBirdclawHelper(); return [ { platform: "contacts", @@ -110,5 +112,27 @@ export function buildLocalIntegrationStates(): ManagedIntegrationState[] { importedFrom: "local-system", artifactPaths: existsSync(chatDbPath) ? [chatDbPath] : [], }, + { + platform: "x", + accountKey: "default", + displayName: "X Archive", + authState: birdclaw.available + ? "authorized" + : birdclaw.helperPath + ? "missing" + : "native_helper_missing", + enabled: true, + connectionKind: "local-cli", + runtimeKind: "native", + syncCapable: birdclaw.available, + importedFrom: "x-archive", + artifactPaths: + birdclaw.archivePath && existsSync(birdclaw.archivePath) ? [birdclaw.archivePath] : [], + metadata: { + helperPath: birdclaw.helperPath, + archivePath: birdclaw.archivePath, + reason: birdclaw.reason, + }, + }, ]; } diff --git a/src/platforms/core/state/status.ts b/src/platforms/core/state/status.ts index f54668ff..c1cc86ce 100644 --- a/src/platforms/core/state/status.ts +++ b/src/platforms/core/state/status.ts @@ -704,6 +704,7 @@ function buildSetupIntegrations( "linkedin", "whatsapp", "signal", + "x", ]; const byPlatform = new Map(); for (const integration of existingIntegrations ?? listIntegrationStates(db, options)) { diff --git a/src/platforms/core/types.ts b/src/platforms/core/types.ts index fd30efc2..f5d7478c 100644 --- a/src/platforms/core/types.ts +++ b/src/platforms/core/types.ts @@ -7,6 +7,7 @@ export const PLATFORM_VALUES = [ "signal", "slack", "whatsapp", + "x", ] as const; export type Platform = (typeof PLATFORM_VALUES)[number]; @@ -17,6 +18,7 @@ export const PLATFORM_PERMISSION_REQUIREMENT_VALUES = ["contacts", "full_disk_ac export type PlatformPermissionRequirement = (typeof PLATFORM_PERMISSION_REQUIREMENT_VALUES)[number]; export const PLATFORM_HELPER_REQUIREMENT_VALUES = [ + "x_archive_helper", "signal_cli", "slack_helper", "whatsapp_helper", @@ -141,6 +143,17 @@ export const PLATFORM_DEFINITIONS = { permissionRequirements: [], helperRequirements: ["whatsapp_helper"], }, + x: { + adapter: true, + defaultAccountKey: "default", + supportsMultipleAccounts: false, + requestableIntegration: false, + requestableOrder: 7, + supportedHostOs: ["macos"], + onboardingVisible: true, + permissionRequirements: [], + helperRequirements: ["x_archive_helper"], + }, } as const satisfies Record; export const PLATFORM_FEATURE_MATRIX = { @@ -240,6 +253,18 @@ export const PLATFORM_FEATURE_MATRIX = { attachments: "yes", contact_sync: "partial", }, + x: { + receive: "yes", + realtime_ingest: "no", + full_history_sync: "yes", + message_edits: "no", + deletes: "no", + reactions: "no", + threads_replies: "no", + read_receipts: "no", + attachments: "no", + contact_sync: "yes", + }, } as const satisfies Record>; type PlatformCapabilityFlag = "adapter" | "requestableIntegration"; diff --git a/src/platforms/x/e2e.test.ts b/src/platforms/x/e2e.test.ts new file mode 100644 index 00000000..edc6515c --- /dev/null +++ b/src/platforms/x/e2e.test.ts @@ -0,0 +1,201 @@ +import { execFileSync } from "node:child_process"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { CuedDatabase } from "../../db/database.js"; +import { projectPendingRawEvents } from "../../runtime/projection/projector.js"; +import { buildXSyncBundle } from "./sync/bundle.js"; + +function writeArchive( + root: string, + input: { + name: string; + follower: boolean; + groupName: string; + groupMember: boolean; + graphOnly: boolean; + }, +): string { + const source = join(root, "archive-source"); + rmSync(source, { recursive: true, force: true }); + mkdirSync(join(source, "data"), { recursive: true }); + writeFileSync( + join(source, "data/account.js"), + `window.YTD.account.part0 = [{"account":{"accountId":"10","username":"sam","accountDisplayName":"Sam"}}]`, + ); + writeFileSync( + join(source, "data/tweets.js"), + `window.YTD.tweets.part0 = [{"tweet":{"entities":{"user_mentions":[{"id_str":"20","screen_name":"ava","name":"${input.name}"}]}}}]`, + ); + writeFileSync( + join(source, "data/direct-messages.js"), + `window.YTD.direct_messages.part0 = [{"dmConversation":{"conversationId":"10-20","messages":[{"messageCreate":{"id":"message-1","senderId":"20","recipientId":"10","text":"Hello from the X archive","createdAt":"2026-08-03T12:00:00.000Z"}}]}}]`, + ); + writeFileSync( + join(source, "data/direct-messages-group.js"), + `window.YTD.direct_messages_group.part0 = [{"dmConversation":{"conversationId":"group-1","name":"${input.groupName}","messages":[${input.groupMember ? "" : '{"participantsLeave":{"initiatingUserId":"30","userIds":["30"],"createdAt":"2026-08-03T14:00:00.000Z"}},'}{"messageCreate":{"id":"group-message-1","senderId":"20","text":"Hello group","createdAt":"2026-08-03T13:00:00.000Z"}},{"joinConversation":{"participantsSnapshot":["10","20","30"],"createdAt":"2026-08-03T11:00:00.000Z"}}]}}]`, + ); + writeFileSync( + join(source, "data/follower.js"), + input.follower + ? `window.YTD.follower.part0 = [{"follower":{"accountId":"20"}}${input.graphOnly ? ',{"follower":{"accountId":"40"}}' : ""}]` + : `window.YTD.follower.part0 = []`, + ); + writeFileSync( + join(source, "data/following.js"), + `window.YTD.following.part0 = [{"following":{"accountId":"20"}}${input.graphOnly ? ',{"following":{"accountId":"40"}}' : ""}]`, + ); + const archivePath = join(root, "twitter-archive.zip"); + rmSync(archivePath, { force: true }); + execFileSync("zip", ["-q", "-r", archivePath, "data"], { cwd: source }); + return archivePath; +} + +async function projectArchiveCycle(db: CuedDatabase) { + let cursor: unknown = null; + let totalApplied = 0; + for (let page = 0; page < 20; page += 1) { + const bundle = await buildXSyncBundle({}, { sourceCursor: cursor }); + db.upsertSourceAccounts(bundle.sourceAccounts ?? []); + db.insertRawEvents(bundle.rawEvents); + totalApplied += projectPendingRawEvents(db).appliedRawEvents; + cursor = bundle.sourceCursor; + if (!bundle.hasMore) return { totalApplied, bundle }; + } + throw new Error("X archive pagination did not complete"); +} + +describe("X archive end to end", () => { + const tempDirs: string[] = []; + + afterEach(() => { + while (tempDirs.length > 0) { + const dir = tempDirs.pop(); + if (dir) rmSync(dir, { recursive: true, force: true }); + } + }); + + it("imports an X archive through the compiled Go helper and refreshes contact state", async () => { + const temp = mkdtempSync(join(tmpdir(), "cued-x-archive-e2e-")); + tempDirs.push(temp); + const archivePath = writeArchive(temp, { + name: "Ava Chen", + follower: true, + groupName: "Builders", + groupMember: true, + graphOnly: true, + }); + const cuedPath = join(temp, "cued.sqlite"); + const helperPath = join(temp, "cued-birdclaw-helper"); + execFileSync("go", ["build", "-o", helperPath, "."], { + cwd: resolve("native/helpers/birdclaw-go"), + env: { ...process.env, GOWORK: "off" }, + }); + + const previousHelper = process.env.CUED_X_ARCHIVE_HELPER_BINARY; + const previousArchive = process.env.CUED_X_ARCHIVE_PATH; + process.env.CUED_X_ARCHIVE_HELPER_BINARY = helperPath; + process.env.CUED_X_ARCHIVE_PATH = archivePath; + const cued = new CuedDatabase(cuedPath); + try { + cued.migrate(); + const first = await projectArchiveCycle(cued); + expect(first.totalApplied).toBeGreaterThanOrEqual(4); + const sqlite = ( + cued as unknown as { + sqlite: { + prepare: (sql: string) => { get: (...params: unknown[]) => Record }; + }; + } + ).sqlite; + const readContact = () => + sqlite + .prepare(` + SELECT c.name, cs.metadata_json + FROM contacts c JOIN contact_sources cs ON cs.contact_id = c.id + WHERE cs.platform = 'x' AND cs.source_entity_key = 'x:user:20' + `) + .get() as { name: string; metadata_json: string }; + expect(readContact().name).toBe("Ava Chen"); + expect(JSON.parse(readContact().metadata_json)).toMatchObject({ + mutual: true, + followsYou: true, + youFollow: true, + }); + expect( + sqlite + .prepare( + "SELECT content FROM messages WHERE platform = 'x' AND content = 'Hello from the X archive'", + ) + .get(), + ).toMatchObject({ content: "Hello from the X archive" }); + const readGraphOnly = () => + sqlite + .prepare( + `SELECT metadata_json, last_seen_at FROM contact_sources WHERE platform = 'x' AND source_entity_key = 'x:user:40'`, + ) + .get() as { metadata_json: string; last_seen_at: number }; + expect(JSON.parse(readGraphOnly().metadata_json)).toMatchObject({ mutual: true }); + const graphOnlyLastSeenAt = readGraphOnly().last_seen_at; + expect( + sqlite + .prepare( + `SELECT COUNT(*) AS count FROM contact_handles WHERE type = 'x_handle' AND value = 'id40'`, + ) + .get(), + ).toMatchObject({ count: 0 }); + const groupBefore = sqlite + .prepare( + `SELECT id, name FROM conversations WHERE platform = 'x' AND source_conversation_key = 'x:conversation:group-1'`, + ) + .get() as { id: string; name: string }; + expect(groupBefore.name).toBe("Builders"); + expect( + sqlite + .prepare( + `SELECT name FROM conversations WHERE platform = 'x' AND source_conversation_key = 'x:conversation:10-20'`, + ) + .get(), + ).toMatchObject({ name: "Ava Chen" }); + + writeArchive(temp, { + name: "Ava Patel", + follower: false, + groupName: "Core team", + groupMember: false, + graphOnly: false, + }); + const second = await projectArchiveCycle(cued); + expect(second.totalApplied).toBeGreaterThan(0); + expect(readContact().name).toBe("Ava Patel"); + expect(JSON.parse(readContact().metadata_json)).toMatchObject({ + mutual: false, + followsYou: false, + youFollow: true, + }); + expect(JSON.parse(readGraphOnly().metadata_json)).toMatchObject({ + mutual: false, + followsYou: false, + youFollow: false, + }); + expect(readGraphOnly().last_seen_at).toBe(graphOnlyLastSeenAt); + expect( + sqlite.prepare(`SELECT name FROM conversations WHERE id = ?`).get(groupBefore.id), + ).toMatchObject({ name: "Core team" }); + expect( + sqlite + .prepare( + `SELECT is_active FROM conversation_participants WHERE conversation_id = ? AND source_participant_key = 'x:user:30'`, + ) + .get(groupBefore.id), + ).toMatchObject({ is_active: 0 }); + } finally { + if (previousHelper === undefined) delete process.env.CUED_X_ARCHIVE_HELPER_BINARY; + else process.env.CUED_X_ARCHIVE_HELPER_BINARY = previousHelper; + if (previousArchive === undefined) delete process.env.CUED_X_ARCHIVE_PATH; + else process.env.CUED_X_ARCHIVE_PATH = previousArchive; + cued.close(); + } + }, 30_000); +}); diff --git a/src/platforms/x/helper/binary.ts b/src/platforms/x/helper/binary.ts new file mode 100644 index 00000000..a7651e3b --- /dev/null +++ b/src/platforms/x/helper/binary.ts @@ -0,0 +1,90 @@ +import { execFileSync } from "node:child_process"; +import { existsSync, readdirSync, statSync } from "node:fs"; +import { homedir } from "node:os"; +import { dirname, join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +const BINARY_NAME = "cued-birdclaw-helper"; +const SUPPORTED_PROTOCOL_VERSION = 1; + +function repoRoot(): string { + return resolve(dirname(fileURLToPath(import.meta.url)), "../../../.."); +} + +export function resolveXArchivePath(env: NodeJS.ProcessEnv = process.env): string | null { + const configured = env.CUED_X_ARCHIVE_PATH?.trim(); + if (configured) return configured; + const downloads = join(homedir(), "Downloads"); + if (!existsSync(downloads)) return null; + return ( + readdirSync(downloads) + .filter((name) => /^(?:twitter.*|x.*archive).*\.zip$/i.test(name)) + .map((name) => join(downloads, name)) + .sort((left, right) => statSync(right).mtimeMs - statSync(left).mtimeMs)[0] ?? null + ); +} + +export function birdclawHelperCandidates(root = repoRoot()): string[] { + return [ + process.env.CUED_APP_PATH?.trim() + ? join(process.env.CUED_APP_PATH.trim(), "Contents", "Resources", "helpers", BINARY_NAME) + : null, + resolve(dirname(fileURLToPath(import.meta.url)), "../../../../../helpers", BINARY_NAME), + join(root, "native", "helpers", "birdclaw-go", ".build", BINARY_NAME), + join(root, "native", "helpers", "birdclaw-go", BINARY_NAME), + ].filter((value): value is string => Boolean(value)); +} + +export function resolveBirdclawHelperBinary( + envValue = process.env.CUED_X_ARCHIVE_HELPER_BINARY, +): string | null { + return ( + envValue?.trim() || + birdclawHelperCandidates().find((candidate) => existsSync(candidate)) || + null + ); +} + +export function inspectBirdclawHelper(): { + helperPath: string | null; + archivePath: string | null; + available: boolean; + reason: string | null; +} { + const helperPath = resolveBirdclawHelperBinary(); + const archivePath = resolveXArchivePath(); + if (!helperPath) { + return { + helperPath: null, + archivePath, + available: false, + reason: "X archive helper is missing", + }; + } + if (!archivePath || !existsSync(archivePath)) { + return { helperPath, archivePath, available: false, reason: "X archive is missing" }; + } + try { + const output = execFileSync(helperPath, ["status", "--archive", archivePath], { + encoding: "utf8", + stdio: ["ignore", "pipe", "pipe"], + }); + const parsed = JSON.parse(output) as { protocolVersion?: unknown; available?: unknown }; + if (parsed.protocolVersion !== SUPPORTED_PROTOCOL_VERSION || parsed.available !== true) { + return { + helperPath, + archivePath, + available: false, + reason: `Unsupported X archive helper protocol: ${String(parsed.protocolVersion)}`, + }; + } + return { helperPath, archivePath, available: true, reason: null }; + } catch (error) { + return { + helperPath, + archivePath, + available: false, + reason: error instanceof Error ? error.message : String(error), + }; + } +} diff --git a/src/platforms/x/sync/bundle.test.ts b/src/platforms/x/sync/bundle.test.ts new file mode 100644 index 00000000..4266ed55 --- /dev/null +++ b/src/platforms/x/sync/bundle.test.ts @@ -0,0 +1,185 @@ +import { describe, expect, it } from "vitest"; +import { type BirdclawSnapshot, buildXSyncBundle } from "./bundle.js"; + +const snapshot: BirdclawSnapshot = { + generation: "archive-v1", + account: { id: "10", name: "Sam", handle: "sam", external_user_id: "10" }, + profiles: [ + { + id: "profile_user_20", + handle: "ava", + display_name: "Ava", + bio: "Robotics", + followers_count: 100, + following_count: 80, + avatar_url: "https://pbs.twimg.com/ava.jpg", + location: "New York", + url: "https://ava.example", + verified_type: "blue", + }, + ], + conversations: [ + { + id: "10-20", + participant_ids: ["20"], + title: "Ava", + last_message_at: "2026-08-03T12:00:00.000Z", + is_group: false, + }, + ], + messages: [ + { + id: "message-1", + conversation_id: "10-20", + sender_profile_id: "profile_user_20", + text: "Hello from Birdclaw", + created_at: "2026-08-03T12:00:00.000Z", + direction: "inbound", + }, + ], + followEdges: [ + { + direction: "followers", + profile_id: "profile_user_20", + external_user_id: "20", + source: "archive", + first_seen_at: "2026-01-01T00:00:00.000Z", + last_seen_at: "2026-08-01T00:00:00.000Z", + }, + { + direction: "following", + profile_id: "profile_user_20", + external_user_id: "20", + source: "archive", + first_seen_at: "2026-01-01T00:00:00.000Z", + last_seen_at: "2026-08-01T00:00:00.000Z", + }, + ], + hasMore: false, + nextMessageOffset: 1, + totalMessageCount: 1, + totalRelationshipCount: 2, +}; + +describe("Birdclaw X sync bundle", () => { + it("normalizes DMs, deterministic X identities, and mutual graph metadata", async () => { + const bundle = await buildXSyncBundle({ accountKey: "default" }, { snapshot }); + + expect(bundle.sourceAccounts).toEqual([ + { platform: "x", accountKey: "10", displayName: "@sam" }, + ]); + expect(bundle.rawEvents.map((event) => `${event.entityKind}.${event.eventKind}`)).toEqual([ + "contact.observed", + "conversation.observed", + "message.created", + "contact.snapshot_completed", + ]); + expect(bundle.rawEvents[0]?.payload).toMatchObject({ + sourceEntityKey: "x:user:20", + handles: expect.arrayContaining([ + { type: "x_user_id", value: "20", deterministic: true }, + { type: "x_handle", value: "ava", deterministic: true }, + ]), + sourceMetadata: { + archiveGeneration: "archive-v1", + followsYou: true, + youFollow: true, + mutual: true, + followersCount: 100, + }, + }); + expect(bundle.rawEvents.find((event) => event.entityKind === "message")?.payload).toMatchObject( + { + content: "Hello from Birdclaw", + senderSourceKey: "x:user:20", + isFromMe: false, + }, + ); + expect(bundle.proofs?.[0]?.coverage).toMatchObject({ + source: "x_archive", + generation: "archive-v1", + relationshipEdges: 2, + }); + expect(bundle.proofs?.[0]?.stats).toMatchObject({ + messageCount: 1, + relationshipEdgeCount: 2, + pageRawEventCount: 4, + }); + }); + + it("continues from the helper cursor when another DM page remains", async () => { + const bundle = await buildXSyncBundle( + {}, + { + snapshot: { + ...snapshot, + hasMore: true, + nextPhase: "relationships", + nextRelationshipUserId: "20", + }, + sourceCursor: { messageOffset: 0 }, + }, + ); + expect(bundle.hasMore).toBe(true); + expect(bundle.sourceCursor).toEqual({ + generation: "archive-v1", + phase: "relationships", + messageOffset: 1, + relationshipUserId: "20", + totalMessageCount: 1, + totalRelationshipCount: 2, + }); + expect(bundle.proofs?.[0]?.status).toBe("running"); + }); + + it("keeps outbound and group participants on real X user identities", async () => { + const bundle = await buildXSyncBundle( + {}, + { + snapshot: { + ...snapshot, + profiles: [ + ...snapshot.profiles, + { ...snapshot.profiles[0]!, id: "profile_user_30", handle: "lee", display_name: "Lee" }, + ], + conversations: [ + { + id: "group-1", + participant_ids: ["20", "30"], + title: "Builders", + last_message_at: "2026-08-03T13:00:00.000Z", + is_group: true, + }, + ], + messages: [ + { + id: "message-2", + conversation_id: "group-1", + sender_profile_id: "profile_user_10", + text: "Sent by me", + created_at: "2026-08-03T13:00:00.000Z", + direction: "outbound", + }, + ], + followEdges: [], + }, + }, + ); + const conversation = bundle.rawEvents.find((event) => event.entityKind === "conversation"); + expect(conversation?.payload).toMatchObject({ + conversationType: "group", + participantsAreSnapshot: true, + participants: [ + { sourceEntityKey: "x:user:10", isSelf: true }, + { sourceEntityKey: "x:user:20" }, + { sourceEntityKey: "x:user:30" }, + ], + }); + expect(bundle.rawEvents.find((event) => event.entityKind === "message")?.payload).toMatchObject( + { + senderSourceKey: "x:user:10", + isFromMe: true, + }, + ); + }); +}); diff --git a/src/platforms/x/sync/bundle.ts b/src/platforms/x/sync/bundle.ts new file mode 100644 index 00000000..b621f992 --- /dev/null +++ b/src/platforms/x/sync/bundle.ts @@ -0,0 +1,351 @@ +import { execFile } from "node:child_process"; +import { createHash } from "node:crypto"; +import { promisify } from "node:util"; +import type { + ContactObservationPayload, + ContactSnapshotCompletedPayload, + ConversationObservationPayload, + MessagePayload, + ProviderRawEventInput, +} from "../../../core/types/provider.js"; +import type { SyncBundle } from "../../core/sync.js"; +import { resolveBirdclawHelperBinary, resolveXArchivePath } from "../helper/binary.js"; + +const execFileAsync = promisify(execFile); + +export interface BirdclawSnapshot { + generation: string; + account: { id: string; name: string; handle: string; external_user_id: string }; + profiles: Array<{ + id: string; + handle: string; + display_name: string; + bio: string; + followers_count: number; + following_count: number; + avatar_url: string; + location: string; + url: string; + verified_type: string; + }>; + conversations: Array<{ + id: string; + participant_ids: string[]; + title: string; + last_message_at: string; + is_group: boolean; + }>; + messages: Array<{ + id: string; + conversation_id: string; + sender_profile_id: string; + text: string; + created_at: string; + direction: string; + }>; + followEdges: Array<{ + direction: string; + profile_id: string; + external_user_id: string; + source: string; + first_seen_at: string; + last_seen_at: string; + }>; + hasMore: boolean; + nextPhase?: string; + nextRelationshipUserId?: string; + nextMessageOffset?: number; + totalMessageCount?: number; + totalRelationshipCount?: number; +} + +type BirdclawCursor = { + generation: string | null; + phase: "messages" | "relationships"; + messageOffset: number; + relationshipUserId: string | null; + totalMessageCount: number | null; + totalRelationshipCount: number | null; +}; + +function stableId(seed: string): string { + return createHash("sha256").update(seed).digest("hex"); +} + +function parseCursor(value: unknown): BirdclawCursor { + const cursor = value && typeof value === "object" ? (value as Record) : {}; + return { + generation: typeof cursor.generation === "string" ? cursor.generation : null, + phase: cursor.phase === "relationships" ? "relationships" : "messages", + messageOffset: typeof cursor.messageOffset === "number" ? cursor.messageOffset : 0, + relationshipUserId: + typeof cursor.relationshipUserId === "string" ? cursor.relationshipUserId : null, + totalMessageCount: + typeof cursor.totalMessageCount === "number" ? cursor.totalMessageCount : null, + totalRelationshipCount: + typeof cursor.totalRelationshipCount === "number" ? cursor.totalRelationshipCount : null, + }; +} + +async function readBirdclawSnapshot(cursor: BirdclawCursor): Promise { + const helper = resolveBirdclawHelperBinary(); + const archivePath = resolveXArchivePath(); + if (!helper) throw new Error("X archive helper is not installed"); + if (!archivePath) throw new Error("X archive was not found; set CUED_X_ARCHIVE_PATH"); + const args = [ + "sync", + "--archive", + archivePath, + "--phase", + cursor.phase, + "--message-offset", + String(cursor.messageOffset), + ]; + if (cursor.relationshipUserId) { + args.push("--after-relationship-user-id", cursor.relationshipUserId); + } + if (cursor.generation) args.push("--generation", cursor.generation); + if (cursor.totalMessageCount !== null) { + args.push("--known-message-count", String(cursor.totalMessageCount)); + } + if (cursor.totalRelationshipCount !== null) { + args.push("--known-relationship-count", String(cursor.totalRelationshipCount)); + } + const { stdout } = await execFileAsync(helper, args, { + timeout: 120_000, + maxBuffer: 256 * 1024 * 1024, + }); + return JSON.parse(stdout) as BirdclawSnapshot; +} + +function sourceKey(id: string): string { + return `x:user:${id}`; +} + +function externalUserId(profileId: string): string | null { + return profileId.startsWith("profile_user_") ? profileId.slice("profile_user_".length) : null; +} + +export function buildBirdclawRawEvents(input: { + accountKey: string; + snapshot: BirdclawSnapshot; + observedAt: number; +}): SyncBundle["rawEvents"] { + const { snapshot } = input; + const relationshipByProfile = new Map>(); + for (const edge of snapshot.followEdges ?? []) { + const directions = relationshipByProfile.get(edge.profile_id) ?? new Set(); + directions.add(edge.direction); + relationshipByProfile.set(edge.profile_id, directions); + } + const rawEvents: SyncBundle["rawEvents"] = []; + for (const profile of snapshot.profiles ?? []) { + const externalId = externalUserId(profile.id); + if (!externalId) continue; + const contactSourceKey = sourceKey(externalId); + const directions = relationshipByProfile.get(profile.id) ?? new Set(); + const sourceMetadata = { + archiveGeneration: snapshot.generation, + bio: profile.bio, + location: profile.location, + url: profile.url, + verifiedType: profile.verified_type, + followersCount: profile.followers_count, + followingCount: profile.following_count, + followsYou: directions.has("followers"), + youFollow: directions.has("following"), + mutual: directions.has("followers") && directions.has("following"), + }; + const eventVersion = stableId( + JSON.stringify({ + displayName: profile.display_name, + handle: profile.handle, + avatarUrl: profile.avatar_url, + sourceMetadata, + }), + ); + const id = stableId(`x:contact:${input.accountKey}:${externalId}:${eventVersion}`); + rawEvents.push({ + id, + platform: "x", + accountKey: input.accountKey, + entityKind: "contact", + eventKind: "observed", + externalEntityId: externalId, + observedAt: input.observedAt, + dedupeKey: id, + payload: { + sourceEntityKey: contactSourceKey, + fields: { + display_name: + profile.display_name || (profile.handle ? `@${profile.handle}` : profile.id), + photo_url: profile.avatar_url || null, + }, + handles: [ + ...(externalId !== profile.id + ? [{ type: "x_user_id", value: externalId, deterministic: true }] + : []), + ...(profile.handle + ? [ + { type: "x_handle", value: profile.handle, deterministic: true }, + { + type: "x_profile_url", + value: `https://x.com/${profile.handle}`, + deterministic: true, + }, + ] + : []), + ], + sourceProfileUrl: profile.handle ? `https://x.com/${profile.handle}` : null, + sourceMetadata: { + ...sourceMetadata, + }, + } satisfies ContactObservationPayload, + } satisfies ProviderRawEventInput); + } + + const selfKey = sourceKey(snapshot.account.external_user_id || snapshot.account.id); + for (const conversation of snapshot.conversations ?? []) { + const remoteKeys = conversation.participant_ids.map(sourceKey); + const conversationVersion = stableId( + JSON.stringify({ + generation: snapshot.generation, + isGroup: conversation.is_group, + participantIds: conversation.participant_ids, + title: conversation.title, + }), + ); + const conversationId = stableId( + `x:conversation:${input.accountKey}:${conversation.id}:${conversationVersion}`, + ); + rawEvents.push({ + id: conversationId, + platform: "x", + accountKey: input.accountKey, + entityKind: "conversation", + eventKind: "observed", + conversationExternalId: conversation.id, + observedAt: input.observedAt, + dedupeKey: conversationId, + payload: { + sourceConversationKey: `x:conversation:${conversation.id}`, + conversationType: conversation.is_group ? "group" : "dm", + displayName: conversation.title || "X conversation", + participantsAreSnapshot: true, + participants: [ + { sourceEntityKey: selfKey, isSelf: true }, + ...remoteKeys.map((participantKey) => ({ sourceEntityKey: participantKey })), + ], + } satisfies ConversationObservationPayload, + }); + } + for (const message of snapshot.messages ?? []) { + const sentAt = Date.parse(message.created_at); + const id = stableId(`x:message:${input.accountKey}:${message.id}`); + rawEvents.push({ + id, + platform: "x", + accountKey: input.accountKey, + entityKind: "message", + eventKind: "created", + externalEntityId: message.id, + conversationExternalId: message.conversation_id, + occurredAt: Number.isFinite(sentAt) ? sentAt : input.observedAt, + observedAt: input.observedAt, + dedupeKey: id, + payload: { + sourceMessageKey: `x:message:${message.id}`, + sourceConversationKey: `x:conversation:${message.conversation_id}`, + senderSourceKey: + message.direction === "outbound" + ? selfKey + : externalUserId(message.sender_profile_id) + ? sourceKey(externalUserId(message.sender_profile_id)!) + : null, + sentAt: Number.isFinite(sentAt) ? sentAt : input.observedAt, + content: message.text, + isFromMe: message.direction === "outbound", + } satisfies MessagePayload, + }); + } + if (!snapshot.hasMore) { + const id = stableId(`x:contact-snapshot:${input.accountKey}:${snapshot.generation}`); + rawEvents.push({ + id, + platform: "x", + accountKey: input.accountKey, + entityKind: "contact", + eventKind: "snapshot_completed", + observedAt: input.observedAt, + dedupeKey: id, + payload: { generation: snapshot.generation } satisfies ContactSnapshotCompletedPayload, + } satisfies ProviderRawEventInput); + } + return rawEvents; +} + +export async function buildXSyncBundle( + _input: { accountKey?: string } = {}, + options: { snapshot?: BirdclawSnapshot; sourceCursor?: unknown } = {}, +): Promise { + const cursor = parseCursor(options.sourceCursor); + const snapshot = options.snapshot ?? (await readBirdclawSnapshot(cursor)); + const accountKey = snapshot.account.id; + const observedAt = Date.now(); + const rawEvents = buildBirdclawRawEvents({ accountKey, snapshot, observedAt }); + const sourceCursor: BirdclawCursor = snapshot.hasMore + ? { + generation: snapshot.generation, + phase: snapshot.nextPhase === "relationships" ? "relationships" : "messages", + messageOffset: snapshot.nextMessageOffset ?? cursor.messageOffset, + relationshipUserId: snapshot.nextRelationshipUserId ?? cursor.relationshipUserId, + totalMessageCount: snapshot.totalMessageCount ?? cursor.totalMessageCount, + totalRelationshipCount: snapshot.totalRelationshipCount ?? cursor.totalRelationshipCount, + } + : { + generation: snapshot.generation, + phase: "messages", + messageOffset: 0, + relationshipUserId: null, + totalMessageCount: snapshot.totalMessageCount ?? null, + totalRelationshipCount: snapshot.totalRelationshipCount ?? null, + }; + return { + sourceAccounts: [ + { + platform: "x", + accountKey, + displayName: snapshot.account.handle + ? `@${snapshot.account.handle}` + : snapshot.account.name, + }, + ], + rawEvents, + sourceCursor, + syncMode: "full", + hasMore: snapshot.hasMore, + continuation: snapshot.hasMore + ? { reason: "account_pagination", detail: "X archive page remains" } + : undefined, + proofs: [ + { + scope: { kind: "account", key: "x_archive" }, + proofKind: "messages", + status: snapshot.hasMore ? "running" : "complete", + observedAt, + resumeCursor: snapshot.hasMore ? sourceCursor : null, + coverage: { + source: "x_archive", + generation: snapshot.generation, + relationshipEdges: snapshot.totalRelationshipCount ?? snapshot.followEdges?.length ?? 0, + }, + stats: { + messageCount: snapshot.totalMessageCount ?? snapshot.messages?.length ?? 0, + relationshipEdgeCount: + snapshot.totalRelationshipCount ?? snapshot.followEdges?.length ?? 0, + pageRawEventCount: rawEvents.length, + }, + }, + ], + }; +} diff --git a/src/platforms/x/sync/worker.ts b/src/platforms/x/sync/worker.ts new file mode 100644 index 00000000..34b6cad8 --- /dev/null +++ b/src/platforms/x/sync/worker.ts @@ -0,0 +1,20 @@ +import { readAdapterInvocationEnv } from "../../core/invocation.js"; +import { buildXSyncBundle } from "./bundle.js"; + +async function main(): Promise { + try { + const invocation = readAdapterInvocationEnv("x"); + const bundle = await buildXSyncBundle( + { accountKey: process.env.CUED_ACCOUNT_KEY }, + { sourceCursor: invocation.sourceCursor }, + ); + process.stdout.write(JSON.stringify({ ok: true, bundle })); + } catch (error) { + process.stdout.write( + JSON.stringify({ ok: false, error: error instanceof Error ? error.message : String(error) }), + ); + process.exitCode = 1; + } +} + +void main(); diff --git a/src/runtime/daemon/server.test.ts b/src/runtime/daemon/server.test.ts index bc723db6..0882097e 100644 --- a/src/runtime/daemon/server.test.ts +++ b/src/runtime/daemon/server.test.ts @@ -348,6 +348,33 @@ describe("interactive auth sessions", () => { }); describe("sync resume targets", () => { + it("does not poll metered X reads unless autosync is explicitly configured", () => { + const previous = process.env.CUED_AUTOSYNC_PLATFORMS; + delete process.env.CUED_AUTOSYNC_PLATFORMS; + try { + expect( + getAutoSyncTargets({ + listEnabledSyncTargets: () => [ + { platform: "x", account_key: "default" }, + { platform: "slack", account_key: "workspace" }, + ], + listIntegrationStates: () => [], + }), + ).toEqual([{ platform: "slack", accountKey: "workspace" }]); + + process.env.CUED_AUTOSYNC_PLATFORMS = "x"; + expect( + getAutoSyncTargets({ + listEnabledSyncTargets: () => [], + listIntegrationStates: () => [], + }), + ).toEqual([{ platform: "x", accountKey: "default" }]); + } finally { + if (previous == null) delete process.env.CUED_AUTOSYNC_PLATFORMS; + else process.env.CUED_AUTOSYNC_PLATFORMS = previous; + } + }); + it("allows autosync to be explicitly disabled", () => { const previous = process.env.CUED_AUTOSYNC_PLATFORMS; process.env.CUED_AUTOSYNC_PLATFORMS = "none"; diff --git a/src/runtime/daemon/server.ts b/src/runtime/daemon/server.ts index a0a35fe0..c4c57d60 100644 --- a/src/runtime/daemon/server.ts +++ b/src/runtime/daemon/server.ts @@ -28,7 +28,11 @@ import { buildAdapterInvocationEnv, selectAdapterInvocationProofs, } from "../../platforms/core/invocation.js"; -import { isAdapterPlatform, listAutoSyncPlatforms } from "../../platforms/core/registry.js"; +import { + getAdapterDefinition, + isAdapterPlatform, + listAutoSyncPlatforms, +} from "../../platforms/core/registry.js"; import { isAdapterWorkerError, runAdapter } from "../../platforms/core/runner.js"; import { loadIntegrationSecret } from "../../platforms/core/secrets/keychain.js"; import { refreshLocalIntegrationStates } from "../../platforms/core/state/local-refresh.js"; @@ -407,8 +411,10 @@ export function getAutoSyncTargets( const enabled = db .listEnabledSyncTargets() - .filter((target): target is { platform: AdapterPlatform; account_key: string } => - isAdapterPlatform(target.platform), + .filter( + (target): target is { platform: AdapterPlatform; account_key: string } => + isAdapterPlatform(target.platform) && + getAdapterDefinition(target.platform)?.autoSync === true, ) .map((target) => ({ platform: target.platform, @@ -5372,7 +5378,8 @@ async function dispatchRequest( wakeIngest: schedulers.wakeIngest, onRuntimeStateChanged: onAuthRuntimeStateChanged, shouldQueueAuthenticatedSync: (platform) => - platform !== "whatsapp" || shouldRunRealtimePlatform("whatsapp"), + getAdapterDefinition(platform)?.autoSync === true && + (platform !== "whatsapp" || shouldRunRealtimePlatform("whatsapp")), emitAuthenticatedHook: async (platform, accountKey) => { await emitAuthenticatedHook(db, platform, accountKey); }, diff --git a/src/runtime/onboarding.test.ts b/src/runtime/onboarding.test.ts index 4ba9ca17..179a2bb3 100644 --- a/src/runtime/onboarding.test.ts +++ b/src/runtime/onboarding.test.ts @@ -123,6 +123,7 @@ describe("onboarding snapshot", () => { "linkedin", "whatsapp", "signal", + "x", ]); db.close(); @@ -157,6 +158,7 @@ describe("onboarding snapshot", () => { "linkedin", "whatsapp", "signal", + "x", ]); db.close(); diff --git a/src/runtime/projection/events.ts b/src/runtime/projection/events.ts index 92a64095..26f76407 100644 --- a/src/runtime/projection/events.ts +++ b/src/runtime/projection/events.ts @@ -19,7 +19,7 @@ export type NormalizedProjectedRawEvent = { }; const CANONICAL_SCHEMA_REGISTRY = { - contact: new Set(["observed"]), + contact: new Set(["observed", "snapshot_completed"]), conversation: new Set(["observed", "removed"]), call: new Set(["observed", "deleted"]), message: new Set(["created", "updated", "deleted", "read_receipt"]), @@ -53,6 +53,9 @@ export function assertCanonicalRawEventPayloadForWrite( case "contact.observed@1": assertStringField(normalizedSchema, event.payload, "sourceEntityKey"); return; + case "contact.snapshot_completed@1": + assertStringField(normalizedSchema, event.payload, "generation"); + return; case "conversation.observed@1": case "conversation.removed@1": assertStringField(normalizedSchema, event.payload, "sourceConversationKey"); diff --git a/src/runtime/projection/projector.ts b/src/runtime/projection/projector.ts index 3b0e1a80..c95ddfab 100644 --- a/src/runtime/projection/projector.ts +++ b/src/runtime/projection/projector.ts @@ -1,8 +1,9 @@ import { createHash } from "node:crypto"; -import { eq, type SQL, sql } from "drizzle-orm"; +import { and, eq, type SQL, sql } from "drizzle-orm"; import type { CallPayload, ContactObservationPayload, + ContactSnapshotCompletedPayload, ConversationObservationPayload, MessagePayload, ParticipantPayload, @@ -1209,7 +1210,7 @@ function projectContactObservation( accountKey: event.account_key, sourceEntityKey: payload.sourceEntityKey, profileUrl: normalizeText(payload.sourceProfileUrl ?? null), - metadataJson: null, + metadataJson: payload.sourceMetadata ? JSON.stringify(payload.sourceMetadata) : null, firstSeenAt: event.observed_at, lastSeenAt: event.observed_at, }) @@ -1218,7 +1219,7 @@ function projectContactObservation( set: { contactId, profileUrl: normalizeText(payload.sourceProfileUrl ?? null), - metadataJson: null, + metadataJson: payload.sourceMetadata ? JSON.stringify(payload.sourceMetadata) : null, lastSeenAt: event.observed_at, }, }) @@ -1325,6 +1326,39 @@ function projectConversationObservation( .run(); } + if (event.event_kind !== "removed" && payload.participantsAreSnapshot) { + const participantKeys = new Set( + payload.participants.map((participant) => participant.sourceEntityKey), + ); + const existingParticipants = conn + .select({ + contactId: conversationParticipants.contactId, + sourceParticipantKey: conversationParticipants.sourceParticipantKey, + }) + .from(conversationParticipants) + .where(eq(conversationParticipants.conversationId, conversationId)) + .all(); + for (const participant of existingParticipants) { + if ( + !participant.sourceParticipantKey || + participantKeys.has(participant.sourceParticipantKey) + ) { + continue; + } + conn + .update(conversationParticipants) + .set({ isActive: 0, leftAt: event.observed_at, updatedAt: event.observed_at }) + .where( + and( + eq(conversationParticipants.conversationId, conversationId), + eq(conversationParticipants.contactId, participant.contactId), + eq(conversationParticipants.sourceParticipantKey, participant.sourceParticipantKey), + ), + ) + .run(); + } + } + for (const participant of payload.participants) { const contactId = resolveOrEnsureContact( conn, @@ -1372,6 +1406,48 @@ function projectConversationObservation( changes.dirtyConversationIds.add(conversationId); } +function projectContactSnapshotCompleted( + conn: LocalDbExecutor, + changes: ProjectionChangeSet, + event: ProjectableRawEvent, +): void { + const payload = JSON.parse(event.payload_json) as ContactSnapshotCompletedPayload; + const sources = conn + .select({ + id: contactSources.id, + contactId: contactSources.contactId, + metadataJson: contactSources.metadataJson, + }) + .from(contactSources) + .where( + and( + eq(contactSources.platform, event.platform), + eq(contactSources.accountKey, event.account_key), + ), + ) + .all(); + for (const source of sources) { + const metadata = source.metadataJson + ? (JSON.parse(source.metadataJson) as Record) + : {}; + if (metadata.archiveGeneration === payload.generation) continue; + conn + .update(contactSources) + .set({ + metadataJson: JSON.stringify({ + ...metadata, + archiveGeneration: payload.generation, + followsYou: false, + youFollow: false, + mutual: false, + }), + }) + .where(eq(contactSources.id, source.id)) + .run(); + changes.dirtyContactIds.add(source.contactId); + } +} + function projectMessageEvent( conn: LocalDbExecutor, cache: ProjectionCache, @@ -2391,7 +2467,11 @@ function projectEventBatch( tx.run(sql.raw(`SAVEPOINT projection_event_${shapedEvent.rawEvent.rowid}`)); try { if (shapedEvent.entity_kind === "contact") { - projectContactObservation(tx, eventCache, eventChanges, shapedEvent); + if (shapedEvent.event_kind === "snapshot_completed") { + projectContactSnapshotCompleted(tx, eventChanges, shapedEvent); + } else { + projectContactObservation(tx, eventCache, eventChanges, shapedEvent); + } } else if (shapedEvent.entity_kind === "conversation") { projectConversationObservation(tx, eventCache, eventChanges, shapedEvent); } else if (shapedEvent.entity_kind === "call") {