fix(connectors): the external feed sync reads membership before it writes (TASK-164)

backend/__tests__/unit/routes/admin.globalIntegrations.test.js

@@ -13,6 +13,8 @@ jest.mock('../../../models/OAuthState', () => ({
 jest.mock('../../../models/Pod', () => ({
   findOne: jest.fn(),
   create: jest.fn(),
+  updateOne: jest.fn(),
+  findById: jest.fn(),
 }));
 
 jest.mock('../../../integrations', () => ({
@@ -61,7 +63,17 @@ function createRes() {
 describe('admin global integrations route', () => {
   beforeEach(() => {
     jest.clearAllMocks();
-    Pod.findOne.mockResolvedValue({ _id: 'pod-global', name: 'Global Social Feed' });
+    Pod.findOne.mockResolvedValue({
+      _id: 'pod-global',
+      name: 'Global Social Feed',
+      members: ['admin-1'],
+    });
+    Pod.updateOne.mockResolvedValue({});
+    Pod.findById.mockResolvedValue({
+      _id: 'pod-global',
+      name: 'Global Social Feed',
+      members: ['admin-1', 'admin-2'],
+    });
     GlobalModelConfigService.getConfig.mockResolvedValue({
       llmService: {
         provider: 'auto',
@@ -263,6 +275,76 @@ describe('admin global integrations route', () => {
     expect(Integration.create).not.toHaveBeenCalled();
   });
 
+  it('lists a second admin in the global pod before their first sync (TASK-164)', async () => {
+    const handler = getRouteHandler('/x', 'post');
+    const req = {
+      userId: 'admin-2',
+      body: {
+        enabled: true,
+        accessToken: 'x-token',
+        username: 'commonly',
+        userId: 'x-user-id',
+      },
+    };
+    const res = createRes();
+    Integration.findOne.mockResolvedValueOnce(null);
+    Integration.create.mockResolvedValueOnce({
+      _id: 'x-int-2',
+      type: 'x',
+      status: 'connected',
+      config: {},
+    });
+
+    await handler(req, res);
+
+    // Only the first requester became a Mongo member (at pod creation), so a
+    // second admin's feed would pause on its first sync without this write.
+    expect(Pod.updateOne).toHaveBeenCalledWith(
+      { _id: 'pod-global' },
+      { $addToSet: { members: 'admin-2' } },
+    );
+    // The pod handed on is the refetched one, so its members match the write.
+    expect(Pod.findById).toHaveBeenCalledWith('pod-global');
+    expect(Integration.create).toHaveBeenCalledWith(expect.objectContaining({
+      createdBy: 'admin-2',
+      podId: 'pod-global',
+    }));
+    expect(res.json).toHaveBeenCalledWith({
+      success: true,
+      integration: expect.objectContaining({ _id: 'x-int-2' }),
+    });
+  });
+
+  it('does not re-add an admin the global pod already lists (control)', async () => {
+    const handler = getRouteHandler('/x', 'post');
+    const req = {
+      userId: 'admin-1',
+      body: {
+        enabled: true,
+        accessToken: 'x-token',
+        username: 'commonly',
+        userId: 'x-user-id',
+      },
+    };
+    const res = createRes();
+    Integration.findOne.mockResolvedValueOnce(null);
+    Integration.create.mockResolvedValueOnce({
+      _id: 'x-int-3',
+      type: 'x',
+      status: 'connected',
+      config: {},
+    });
+
+    await handler(req, res);
+
+    expect(Pod.updateOne).not.toHaveBeenCalled();
+    expect(Pod.findById).not.toHaveBeenCalled();
+    expect(res.json).toHaveBeenCalledWith({
+      success: true,
+      integration: expect.objectContaining({ _id: 'x-int-3' }),
+    });
+  });
+
   it('the admin Instagram save keeps the stored token when the body omits it', async () => {
     const handler = getRouteHandler('/instagram', 'post');
     const req = {

backend/__tests__/unit/services/externalFeedService.test.js

@@ -3,6 +3,10 @@ jest.mock('../../../models/Integration', () => ({
   findByIdAndUpdate: jest.fn(),
 }));
 
+jest.mock('../../../models/Pod', () => ({
+  findById: jest.fn(),
+}));
+
 jest.mock('../../../models/Post', () => {
   const Post = jest.fn(function Post(doc) {
     Object.assign(this, doc);
@@ -26,18 +30,22 @@ jest.mock('../../../services/agentEventService', () => ({
   enqueue: jest.fn(),
 }));
 
+const mongoose = require('mongoose');
 const Integration = require('../../../models/Integration');
+const Pod = require('../../../models/Pod');
 const Post = require('../../../models/Post');
 const { AgentInstallation } = require('../../../models/AgentRegistry');
 const registry = require('../../../integrations');
 const AgentEventService = require('../../../services/agentEventService');
 const externalFeedService = require('../../../services/externalFeedService');
 
 describe('externalFeedService', () => {
-  beforeEach(() => {
-    jest.clearAllMocks();
-    delete process.env.EXTERNAL_FEED_PERSIST_POSTS;
-  });
+  // The real shape: the service reads the pod with `.lean()`, so members and a
+  // lean integration's `createdBy` are ObjectIds, not strings. String members
+  // would let a raw `members.includes(owner)` pass for the shared predicate
+  // (TASK-164 ledger M10 - the fixture, not the code, decides that).
+  const OWNER_ID = new mongoose.Types.ObjectId('507f1f77bcf86cd799439011');
+  const OTHER_ID = new mongoose.Types.ObjectId('507f191e810c19729de860ea');
 
   const mockFindChain = (value) => ({
     select: () => ({
@@ -46,6 +54,45 @@ describe('externalFeedService', () => {
     lean: jest.fn().mockResolvedValue(value),
   });
 
+  const mockPodMembers = (members) => {
+    // A fresh instance carrying the same hex, which is what a lean read of the
+    // pod produces: the pod's ObjectId and the integration's are never the same
+    // reference. Admitting the owner has to be a value comparison, so a raw
+    // `members.includes(owner)` - reference equality - is refused here
+    // (TASK-164 ledger M10).
+    const stored = members.map((member) => (
+      member instanceof mongoose.Types.ObjectId
+        ? new mongoose.Types.ObjectId(member.toHexString())
+        : member
+    ));
+    Pod.findById.mockReturnValue({
+      select: () => ({ lean: jest.fn().mockResolvedValue({ _id: 'pod-1', members: stored }) }),
+    });
+  };
+
+  const feedRow = (over = {}) => ({
+    _id: 'int-1',
+    type: 'x',
+    podId: 'pod-1',
+    status: 'connected',
+    isActive: true,
+    createdBy: OWNER_ID,
+    config: { messageBuffer: [], maxBufferSize: 1000 },
+    ...over,
+  });
+
+  const mockOneFeed = (over = {}) => {
+    Integration.find.mockReturnValue({ lean: jest.fn().mockResolvedValue([feedRow(over)]) });
+  };
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    delete process.env.EXTERNAL_FEED_PERSIST_POSTS;
+    // The default fixture is a pod that lists the integration's owner, so every
+    // pre-existing arm below stays a listed-owner arm (TASK-164).
+    mockPodMembers([OWNER_ID]);
+  });
+
   test('does not persist external feed posts by default and enqueues curator events', async () => {
     Integration.find.mockReturnValue({
       lean: jest.fn().mockResolvedValue([
@@ -55,7 +102,7 @@ describe('externalFeedService', () => {
           podId: 'pod-1',
           status: 'connected',
           isActive: true,
-          createdBy: 'user-1',
+          createdBy: OWNER_ID,
           config: { messageBuffer: [], maxBufferSize: 1000 },
         },
       ]),
@@ -107,6 +154,14 @@ describe('externalFeedService', () => {
       createdPosts: 0,
       curatorEventsEnqueued: 1,
     }));
+    // The agreed control for the membership arms below (wren, TASK-164): a
+    // listed owner still reaches the provider, still buffers, still enqueues.
+    expect(registry.get).toHaveBeenCalledTimes(1);
+    expect(Integration.findByIdAndUpdate).toHaveBeenCalledWith('int-1', {
+      $push: {
+        'config.messageBuffer': expect.objectContaining({ $each: expect.any(Array) }),
+      },
+    });
   });
 
   test('can persist external posts when EXTERNAL_FEED_PERSIST_POSTS=1', async () => {
@@ -119,7 +174,7 @@ describe('externalFeedService', () => {
           podId: 'pod-1',
           status: 'connected',
           isActive: true,
-          createdBy: 'user-1',
+          createdBy: OWNER_ID,
           config: { messageBuffer: [], maxBufferSize: 1000 },
         },
       ]),
@@ -160,7 +215,7 @@ describe('externalFeedService', () => {
           podId: 'pod-1',
           status: 'connected',
           isActive: true,
-          createdBy: 'user-1',
+          createdBy: OWNER_ID,
           config: { messageBuffer: [], maxBufferSize: 1000 },
         },
       ]),
@@ -211,7 +266,7 @@ describe('externalFeedService', () => {
           podId: 'pod-1',
           status: 'connected',
           isActive: true,
-          createdBy: 'user-1',
+          createdBy: OWNER_ID,
           config: { messageBuffer: [], maxBufferSize: 1000 },
         },
       ]),
@@ -241,4 +296,88 @@ describe('externalFeedService', () => {
       }),
     );
   });
+
+  describe('owner membership (TASK-164)', () => {
+    test('a departed owner calls no provider, buffers nothing, and pauses the row with the reason', async () => {
+      mockPodMembers([OTHER_ID]);
+      mockOneFeed();
+
+      const results = await externalFeedService.syncExternalFeeds();
+
+      // No provider call, so no cursor advance either: the whole sync is skipped.
+      expect(registry.get).not.toHaveBeenCalled();
+      expect(Post.insertMany).not.toHaveBeenCalled();
+      expect(AgentEventService.enqueue).not.toHaveBeenCalled();
+
+      const writes = Integration.findByIdAndUpdate.mock.calls;
+      expect(writes).toHaveLength(1);
+      const [, update] = writes[0];
+      expect(writes[0][0]).toBe('int-1');
+      expect(update.$push).toBeUndefined();
+      expect(update.$set).toEqual(expect.objectContaining({
+        status: 'error',
+        errorMessageUserFacing: true,
+      }));
+      expect(update.$set.errorMessage).toContain('no longer a member of the pod');
+      // isActive stays true, so the row stays on the owner's Connectors page;
+      // the pause is expressed as status, and nothing here writes the flag.
+      expect(update.$set.isActive).toBeUndefined();
+      expect(results[0]).toEqual(expect.objectContaining({
+        integrationId: 'int-1',
+        success: false,
+        paused: true,
+        messageCount: 0,
+      }));
+    });
+
+    test('a departed owner writes no posts on the flag-on path either', async () => {
+      process.env.EXTERNAL_FEED_PERSIST_POSTS = '1';
+      mockPodMembers([]);
+      mockOneFeed();
+
+      const results = await externalFeedService.syncExternalFeeds();
+
+      expect(Post.find).not.toHaveBeenCalled();
+      expect(Post.insertMany).not.toHaveBeenCalled();
+      expect(registry.get).not.toHaveBeenCalled();
+      expect(results[0]).toEqual(expect.objectContaining({ paused: true, success: false }));
+    });
+
+    test('a pod that is gone pauses rather than syncing into nothing', async () => {
+      Pod.findById.mockReturnValue({ select: () => ({ lean: jest.fn().mockResolvedValue(null) }) });
+      mockOneFeed();
+
+      const results = await externalFeedService.syncExternalFeeds();
+
+      expect(registry.get).not.toHaveBeenCalled();
+      expect(results[0]).toEqual(expect.objectContaining({ paused: true, success: false }));
+    });
+
+    test('a listed owner is unaffected: provider, buffer and curator events all run', async () => {
+      mockPodMembers([OWNER_ID]);
+      mockOneFeed();
+      registry.get.mockReturnValue({
+        syncRecent: jest.fn().mockResolvedValue({
+          messages: [{
+            externalId: 'x-9',
+            content: 'post nine',
+            timestamp: new Date().toISOString(),
+            authorName: 'author',
+          }],
+          content: 'Synced external feed',
+        }),
+      });
+      Post.find.mockImplementation(() => mockFindChain([]));
+      AgentInstallation.find.mockReturnValue(mockFindChain([]));
+
+      const results = await externalFeedService.syncExternalFeeds();
+
+      expect(registry.get).toHaveBeenCalledWith('x', expect.objectContaining({ _id: 'int-1' }));
+      expect(Integration.findByIdAndUpdate).toHaveBeenCalledWith('int-1', expect.objectContaining({
+        $push: expect.anything(),
+      }));
+      expect(results[0]).toEqual(expect.objectContaining({ success: true, messageCount: 1 }));
+      expect(results[0].paused).toBeUndefined();
+    });
+  });
 });

backend/routes/admin/globalIntegrations.ts

@@ -12,6 +12,8 @@ const Pod = require('../../models/Pod');
 const registry = require('../../integrations');
 const SocialPolicyService = require('../../services/socialPolicyService');
 const GlobalModelConfigService = require('../../services/globalModelConfigService');
+// eslint-disable-next-line global-require
+const { isListedPodMember } = require('../../utils/isPodMember');
 const externalFeedService = require('../../services/externalFeedService');
 
 let PGPod = null;
@@ -115,6 +117,17 @@ const ensureGlobalSocialFeedPod = async (userId: any) => {
       createdBy: userId,
       tags: ['social', 'global', 'feeds'],
     });
+  } else if (!isListedPodMember(globalPod, userId)) {
+    // The requester is about to own a feed integration in this pod, and the
+    // sync refuses to write for an owner the pod does not list (TASK-164).
+    // Only the FIRST requester became a Mongo member (at creation); a second
+    // admin configuring the other feed type was mirrored into PG alone, so
+    // their first sync would pause a supported setup. Mongo `members` is what
+    // the predicate reads and what the pod's own write paths enforce; the PG
+    // mirror follows below. `createdBy` is deliberately untouched — it is the
+    // row's owner, not a membership record.
+    await Pod.updateOne({ _id: globalPod._id }, { $addToSet: { members: userId } });
+    globalPod = await Pod.findById(globalPod._id);
   }
 
   await ensureGlobalPodPostgresSync({ pod: globalPod, userId });