Merge remote-tracking branch 'origin/main'

# Conflicts:
#	src/browser/client.ts
#	src/services/reddit-session-importer.ts
#	src/web/routes/conversations.ts

src/services/account-auth-context.ts

@@ -6,11 +6,18 @@ export type AccountAuthStatus =
   | 'authenticated';
 
 export interface AuthenticatedAccount {
+  /** Reddit account fullname, e.g. `t2_abc123`. */
   accountId: string;
   redditUsername: string;
   matrixUserId: string;
 }
 
+export interface RedditAccountIdentity {
+  username: string;
+  /** Reddit's account ID, with or without the `t2_` prefix. */
+  accountId: string;
+}
+
 export interface AccountAuthSnapshot {
   status: AccountAuthStatus;
   accountId: string | null;
@@ -34,21 +41,43 @@ export interface AccountAuthContextOptions {
 }
 
 export interface CurrentAccountAuthentication {
-  getRedditUsername(): Promise<string | null>;
+  getRedditIdentity(): Promise<RedditAccountIdentity | null>;
   refreshMatrixCredentials(): Promise<{ userId: string } | null>;
 }
 
+export function normalizeRedditAccountId(accountId: string): string {
+  const trimmed = accountId.trim();
+  const normalized = trimmed.startsWith('t2_') ? trimmed : `t2_${trimmed}`;
+  if (!/^t2_[a-z0-9]+$/i.test(normalized)) {
+    throw new Error('Reddit returned an invalid account ID');
+  }
+  return normalized;
+}
+
+export function expectedMatrixUserId(accountId: string): string {
+  return `@${normalizeRedditAccountId(accountId)}:reddit.com`;
+}
+
+export function assertMatchingAccountIdentity(account: AuthenticatedAccount): void {
+  const expected = expectedMatrixUserId(account.accountId);
+  if (account.matrixUserId !== expected) {
+    throw new Error(
+      `Reddit account u/${account.redditUsername} (${normalizeRedditAccountId(account.accountId)}) does not match Matrix account ${account.matrixUserId}`,
+    );
+  }
+}
+
 export async function authenticateCurrentAccount(
   authentication: CurrentAccountAuthentication,
 ): Promise<AuthenticatedAccount | null> {
-  const redditUsername = await authentication.getRedditUsername();
-  if (!redditUsername) return null;
+  const redditIdentity = await authentication.getRedditIdentity();
+  if (!redditIdentity) return null;
 
   const credentials = await authentication.refreshMatrixCredentials();
   if (!credentials?.userId) return null;
   return {
-    accountId: credentials.userId,
-    redditUsername,
+    accountId: normalizeRedditAccountId(redditIdentity.accountId),
+    redditUsername: redditIdentity.username,
     matrixUserId: credentials.userId,
   };
 }
@@ -104,15 +133,15 @@ export class AccountAuthContext implements AccountAuthGate {
     if (!account.accountId || !account.redditUsername || !account.matrixUserId) {
       throw new Error('A fully authenticated account requires Reddit and Matrix identities');
     }
-    if (account.accountId !== account.matrixUserId) {
-      throw new Error('Authenticated account ID must match the Matrix user ID');
-    }
+    assertMatchingAccountIdentity(account);
 
-    const changedAccount = this.snapshot.accountId !== account.accountId;
+    const accountId = normalizeRedditAccountId(account.accountId);
+
+    const changedAccount = this.snapshot.accountId !== accountId;
     const wasAuthenticated = this.isFullyAuthenticated();
     this.snapshot = {
       status: 'authenticated',
-      accountId: account.accountId,
+      accountId,
       redditUsername: account.redditUsername,
       matrixUserId: account.matrixUserId,
       reason: null,

src/services/chat/matrix/client.ts

@@ -54,6 +54,11 @@ export interface RoomSummary {
   recentEvents: MatrixTimelineEvent[];
 }
 
+export interface DirectRoomMapping {
+  recipientMxid: string;
+  roomId: string;
+}
+
 /**
  * Cap on how long we'll sit and sleep waiting out a 429. Anything longer
  * is surfaced as a `MatrixRateLimitError` so the main loop can back off
@@ -405,7 +410,7 @@ export class MatrixClient {
    * "Start chat" in the UI:
    *
    *   POST /_matrix/client/v3/createRoom
-   *   { "preset": "reddit_dm", "invite": [mxid] }
+   *   { "preset": "reddit_dm", "invite": [mxid], "is_direct": true }
    *
    * `reddit_dm` is a Reddit-specific preset that the homeserver expands
    * server-side into the right power levels, history visibility, and
@@ -413,12 +418,53 @@ export class MatrixClient {
    */
   async createDirectRoom(recipientMxid: string): Promise<string> {
     const res = await this.request<{ room_id: string }>('POST', '/_matrix/client/v3/createRoom', {
-      body: { preset: 'reddit_dm', invite: [recipientMxid] },
+      body: { preset: 'reddit_dm', invite: [recipientMxid], is_direct: true },
     });
     this.logger.debug(`[matrix] created DM room ${res.room_id} with ${recipientMxid}`);
     return res.room_id;
   }
 
+  /**
+   * Add missing rooms to the account-level `m.direct` map used by Matrix
+   * clients to classify joined rooms as direct chats. Existing mappings are
+   * preserved verbatim; the write is skipped when there is nothing to repair.
+   */
+  async repairDirectRoomMappings(mappings: DirectRoomMapping[]): Promise<number> {
+    if (mappings.length === 0) return 0;
+
+    let directRooms: Record<string, string[]>;
+    try {
+      directRooms = await this.request<Record<string, string[]>>(
+        'GET',
+        `/_matrix/client/v3/user/${encodeURIComponent(this.creds.userId)}/account_data/m.direct`,
+      );
+    } catch (error) {
+      if (error instanceof Error && error.message.includes('→ 404')) {
+        directRooms = {};
+      } else {
+        throw error;
+      }
+    }
+
+    let repaired = 0;
+    for (const { recipientMxid, roomId } of mappings) {
+      const existing = Array.isArray(directRooms[recipientMxid])
+        ? directRooms[recipientMxid]
+        : [];
+      if (existing.includes(roomId)) continue;
+      directRooms[recipientMxid] = [...existing, roomId];
+      repaired++;
+    }
+
+    if (repaired === 0) return 0;
+    await this.request<Record<string, never>>(
+      'PUT',
+      `/_matrix/client/v3/user/${encodeURIComponent(this.creds.userId)}/account_data/m.direct`,
+      { body: directRooms },
+    );
+    return repaired;
+  }
+
   /**
    * One-shot: create a DM room with the recipient and send the opening
    * message. Returns the new room_id and the event_id of the message.
@@ -428,6 +474,13 @@ export class MatrixClient {
     body: string,
   ): Promise<{ roomId: string; eventId: string }> {
     const roomId = await this.createDirectRoom(recipientMxid);
+    try {
+      await this.repairDirectRoomMappings([{ recipientMxid, roomId }]);
+    } catch (error) {
+      // Room creation has already succeeded. Do not turn a metadata repair
+      // failure into an orphan-and-retry cycle that creates duplicate rooms.
+      this.logger.warn(`[matrix] failed to register ${roomId} as a direct chat:`, error);
+    }
     const eventId = await this.sendTextMessage(roomId, body);
     return { roomId, eventId };
   }

src/services/chat/service.ts

@@ -402,6 +402,7 @@ export class ChatService {
           if (!chat.roomName || chat.roomName.startsWith('t2_')) chat.roomName = resolved;
         }
         await this.cache.save(chats);
+        await this.repairDirectRoomMappings(chats);
 
         let done = 0;
         for (const chat of chats) {
@@ -736,17 +737,21 @@ export class ChatService {
       existing.set(roomId, next);
     }
 
-    if (this.mediaService) {
-      for (const [roomId, room] of Object.entries(joined)) {
-        const scraped = this.scrapeMessageEvents(room?.timeline?.events ?? []);
-        const mediaMessages = scraped.filter((m) => m.mediaUrls.length > 0);
-        if (mediaMessages.length === 0) continue;
-        const chat = existing.get(roomId);
-        if (!chat) continue;
+    for (const [roomId, room] of Object.entries(joined)) {
+      const scraped = this.scrapeMessageEvents(room?.timeline?.events ?? []);
+      if (scraped.length === 0) continue;
+      const chat = existing.get(roomId);
+      if (!chat) continue;
+
+      let newMessages: ChatMessage[] = scraped.map(
+        ({ mediaUrls: _ignored, ...message }) => message,
+      );
+      if (this.mediaService && scraped.some((message) => message.mediaUrls.length > 0)) {
         try {
-          const withMedia = await this.mediaService.attachMedia(chat, mediaMessages);
-          this.mergeNewMessages(chat, withMedia);
-          if (withMedia.some((message) => !message.isSelf && (message.media?.length ?? 0) > 0)) {
+          newMessages = await this.mediaService.attachMedia(chat, scraped);
+          if (newMessages.some(
+            (message) => !message.isSelf && (message.media?.length ?? 0) > 0,
+          )) {
             this.recordReceivedImage(chat);
           }
         } catch (err) {
@@ -755,6 +760,7 @@ export class ChatService {
           );
         }
       }
+      this.mergeNewMessages(chat, newMessages);
     }
 
     const merged = Array.from(existing.values()).sort(
@@ -766,9 +772,49 @@ export class ChatService {
     // Resolve any t2_-prefixed usernames the delta just introduced, so the
     // dashboard doesn't show raw Matrix-style IDs for new rooms.
     await this.enrichUsernames();
+    await this.repairDirectRoomMappings(merged);
     return merged;
   }
 
+  /**
+   * Reconcile cached one-to-one chats with Matrix account data. The durable
+   * username repository lets delta sync repair older rooms even when Reddit
+   * omits their member state from the latest /sync response.
+   */
+  private async repairDirectRoomMappings(chats: ActiveChat[]): Promise<void> {
+    const mappings: Array<{ recipientMxid: string; roomId: string }> = [];
+    const seen = new Set<string>();
+
+    for (const chat of chats) {
+      if (!chat.roomId) continue;
+      const rawT2 = this.extractT2(chat);
+      const persistedT2 = rawT2
+        ? `t2_${rawT2}`
+        : this.redditUsersRepo?.getByUsername(chat.username)?.t2Id;
+      if (!persistedT2) continue;
+
+      const recipientMxid = MatrixClient.t2ToMxid(persistedT2);
+      const key = `${recipientMxid}|${chat.roomId}`;
+      if (seen.has(key)) continue;
+      seen.add(key);
+      mappings.push({ recipientMxid, roomId: chat.roomId });
+    }
+
+    if (mappings.length === 0) return;
+    try {
+      const repaired = await this.withAuth(() =>
+        this.matrix.repairDirectRoomMappings(mappings),
+      );
+      if (repaired > 0) {
+        this.logger.info(`[chat] repaired ${repaired} missing direct-room mapping(s)`);
+      }
+    } catch (error) {
+      // Sync remains useful even if Reddit temporarily rejects account-data
+      // maintenance. A later delta/full sync will try the additive repair again.
+      this.logger.warn('[chat] direct-room mapping repair failed:', error);
+    }
+  }
+
   private scrapeMessageEvents(events: MatrixTimelineEvent[]): ScrapedChatMessage[] {
     return events
       .filter((e) => e.type === 'm.room.message')

src/services/monitor.ts

@@ -36,9 +36,12 @@ export class MonitorService {
         this.logger.debug(`Checking r/${subreddit.name} (${subreddit.type}) for new posts...`);
         const posts = await this.redditApi.getNewPosts(subreddit.name, 25, subreddit.type);
 
-        // Filter for valid, unprocessed posts
-        const validPosts: RedditPost[] = [];
-        for (const post of posts) {
+        // Reddit normally returns /new in descending order, but sort explicitly
+        // so we can stop as soon as the first fully eligible candidate is found.
+        // Previously we screened the entire batch before choosing one, causing up
+        // to two profile requests for every otherwise-valid post in the listing.
+        const newestFirst = [...posts].sort((a, b) => b.created_utc - a.created_utc);
+        for (const post of newestFirst) {
           if (this.postsRepo.isProcessed(post.id)) {
             continue;
           }
@@ -48,6 +51,23 @@ export class MonitorService {
             continue;
           }
 
+          if (
+            this.config.reddit.filterSellers &&
+            (this.config.reddit.sellerKeywords.length > 0 || this.config.reddit.sellerDomains.length > 0)
+          ) {
+            const isSeller = await this.redditApi.isSellerProfile(
+              post.author,
+              this.config.reddit.sellerKeywords,
+              this.config.reddit.sellerDomains,
+              post,
+            );
+            if (isSeller) {
+              this.logger.info(`Skipping post ${post.id} from u/${post.author} - seller signals detected`);
+              this.recordRejection(post, subreddit.type, 'Seller signals detected');
+              continue;
+            }
+          }
+
           // Chat subs almost always have a self-declared gender tag in the
           // title/body (e.g. [25F], M4F). Drop posts that don't match the
           // configured targetGenders here so they never enter the processor.
@@ -55,41 +75,33 @@ export class MonitorService {
             const tag = parseGenderTag(`${post.title}\n${post.selftext ?? ''}`);
             if (!tag) {
               this.logger.debug(`Skipping chat post ${post.id} - no gender tag in title/body`);
+              this.recordRejection(post, subreddit.type, 'No gender tag in title/body');
               continue;
             }
             if (!targetGenders.includes(tag)) {
               this.logger.debug(
                 `Skipping chat post ${post.id} - gender tag "${tag}" not in target list [${targetGenders.join(', ')}]`,
               );
+              this.recordRejection(post, subreddit.type, `Gender tag "${tag}" not targeted`);
               continue;
             }
           }
 
-          validPosts.push(post);
-        }
-
-        if (validPosts.length > 0) {
-          // Sort by creation time (newest first)
-          validPosts.sort((a, b) => b.created_utc - a.created_utc);
-
-          if (prioritizeRecent) {
-            const recent = validPosts.find((p) => p.created_utc >= recentCutoffSec);
-            if (recent) {
-              this.logger.info(`Found ${validPosts.length} new posts in r/${subreddit.name} (recent match)`);
-              return [recent];
-            }
-            // Remember the freshest non-recent candidate as a fallback and
-            // keep scanning other subs for an actually-recent post.
-            if (!fallback || validPosts[0].created_utc > fallback.created_utc) {
-              fallback = validPosts[0];
-            }
-          } else {
-            this.logger.info(`Found ${validPosts.length} new posts in r/${subreddit.name}`);
-            // Return just the first (newest) valid post
-            // This ensures we switch to a different random subreddit on the next poll cycle
-            // instead of processing a batch from the same subreddit
-            return [validPosts[0]];
+          // At this point the candidate has passed every check. Select it now
+          // rather than spending more Reddit requests screening users that
+          // cannot be acted on during this poll cycle.
+          if (!prioritizeRecent || post.created_utc >= recentCutoffSec) {
+            this.logger.info(
+              `Found eligible post in r/${subreddit.name}${prioritizeRecent ? ' (recent match)' : ''}`,
+            );
+            return [post];
           }
+
+          // This is the newest eligible post in this subreddit. Keep it as a
+          // possible fallback, then move to the next subreddit to look for a
+          // recent candidate. Older posts in this batch cannot improve it.
+          if (!fallback || post.created_utc > fallback.created_utc) fallback = post;
+          break;
         }
       } catch (error) {
         if (error instanceof RedditRateLimitError) throw error;
@@ -160,4 +172,22 @@ export class MonitorService {
   private sleep(ms: number): Promise<void> {
     return new Promise((resolve) => setTimeout(resolve, ms));
   }
+
+  /** Persist a completed screening decision so later polls do not repeat it. */
+  private recordRejection(
+    post: RedditPost,
+    subredditType: 'image' | 'chat',
+    reason: string,
+  ): void {
+    this.postsRepo.insert({
+      id: post.id,
+      subreddit: post.subreddit,
+      author: post.author,
+      title: post.title,
+      postUrl: this.redditApi.getPostUrl(post),
+      imageUrl: subredditType === 'image' ? this.redditApi.getImageUrl(post) : null,
+      processedAt: Date.now(),
+    });
+    this.postsRepo.setError(post.id, reason);
+  }
 }

src/services/reddit-session-importer.ts

@@ -2,12 +2,16 @@ import type { Cookie } from 'playwright';
 import type { MatrixClient } from './chat/matrix/client.js';
 import type { MatrixCredentials } from './chat/matrix/types.js';
 import type { AccountAuthContext } from './account-auth-context.js';
+import {
+  assertMatchingAccountIdentity,
+  normalizeRedditAccountId,
+  type RedditAccountIdentity,
+} from './account-auth-context.js';
 
 interface BrowserSessionTarget {
   importRedditCookies(cookies: Cookie[]): Promise<void>;
   isSessionValid(): Promise<boolean>;
-  getAuthenticatedUsername(): Promise<string | null>;
-  getAuthenticatedIdentity?(): Promise<{ username: string; accountId: string } | null>;
+  getAuthenticatedIdentity(): Promise<RedditAccountIdentity | null>;
 }
 
 interface ChatSessionTarget {
@@ -28,11 +32,6 @@ export interface VerifiedRedditSessionStore {
   }): void;
 }
 
-function matrixRedditAccountId(matrixUserId: string): string | null {
-  const match = matrixUserId.match(/^@t2_([^:]+):reddit\.com$/i);
-  return match?.[1]?.toLowerCase() ?? null;
-}
-
 export class RedditSessionImporter {
   constructor(
     private readonly browser: BrowserSessionTarget,
@@ -63,31 +62,37 @@ export class RedditSessionImporter {
     }
 
     await this.browser.importRedditCookies(cookies);
-    if (!(await this.browser.isSessionValid())) {
-      throw new Error('Imported cookies did not authenticate with Reddit');
-    }
-    const identity = this.browser.getAuthenticatedIdentity
-      ? await this.browser.getAuthenticatedIdentity()
-      : null;
-    const redditUsername = identity?.username ?? await this.browser.getAuthenticatedUsername();
-    if (!redditUsername) throw new Error('Reddit did not return an authenticated account identity');
-    if (identity) {
-      const matrixAccountId = matrixRedditAccountId(userId);
-      if (!matrixAccountId || matrixAccountId !== identity.accountId.toLowerCase().replace(/^t2_/, '')) {
-        throw new Error('Reddit cookies and Matrix credentials belong to different accounts');
+    let redditIdentity: RedditAccountIdentity;
+    try {
+      if (!(await this.browser.isSessionValid())) {
+        throw new Error('Imported cookies did not authenticate with Reddit');
       }
+      const identity = await this.browser.getAuthenticatedIdentity();
+      if (!identity) throw new Error('Reddit did not return an authenticated account identity');
+      redditIdentity = identity;
+      assertMatchingAccountIdentity({
+        accountId: identity.accountId,
+        redditUsername: identity.username,
+        matrixUserId: userId,
+      });
+    } catch (error) {
+      this.accountAuth.invalidate(error instanceof Error ? error.message : String(error));
+      throw error;
     }
-
     this.saveCredentials(credentials);
     this.chat.setMatrixClient(matrixClient);
     const account = {
       accountId: userId,
-      redditUsername,
+      redditUsername: redditIdentity.username,
       matrixUserId: userId,
     };
+    this.accountAuth.markAuthenticated({
+      accountId: normalizeRedditAccountId(redditIdentity.accountId),
+      redditUsername: redditIdentity.username,
+      matrixUserId: userId,
+    });
     this.sessionStore?.save({ ...account, cookies, credentials });
     await this.chat.switchAccount?.(account.accountId);
-    this.accountAuth.markAuthenticated(account);
     return account;
   }
 }