Replace webhooks with polling sources for Trustpilot integration

- Replaced webhook approach with 15-minute polling (webhooks not supported by Trustpilot)
- Added conversation endpoints and public review endpoints to app
- Created comprehensive polling base class with deduplication by reviewId + timestamp
- Implemented 8 polling sources as requested:
  - new-service-reviews (public + private)
  - updated-service-reviews
  - new-product-reviews
  - updated-product-reviews
  - new-service-review-replies
  - new-product-review-replies
  - new-conversations
  - updated-conversations
- Added proper business unit filtering and 24-hour lookback on first run
- Skipped deleted reviews (not detectable via polling)
- Maintained all existing actions (no changes needed)

Author: seynadio

components/trustpilot/app/trustpilot.app.ts

@@ -180,7 +180,63 @@ export default defineApp({
       return response.businessUnits?.map(parseBusinessUnit) || [];
     },
 
-    // Service Review methods
+    // Public Review methods (no auth required for basic info)
+    async getPublicServiceReviews({
+      businessUnitId,
+      stars = null,
+      sortBy = SORT_OPTIONS.CREATED_AT_DESC,
+      limit = DEFAULT_LIMIT,
+      offset = 0,
+      tags = [],
+      language = null,
+    } = {}) {
+      if (!validateBusinessUnitId(businessUnitId)) {
+        throw new Error("Invalid business unit ID");
+      }
+
+      const endpoint = buildUrl(ENDPOINTS.PUBLIC_REVIEWS, { businessUnitId });
+      const params = {
+        stars,
+        orderBy: sortBy,
+        perPage: limit,
+        page: Math.floor(offset / limit) + 1,
+        language,
+      };
+
+      if (tags.length > 0) {
+        params.tags = tags.join(",");
+      }
+
+      const response = await this._makeRequestWithRetry({
+        endpoint,
+        params,
+      });
+
+      return {
+        reviews: response.reviews?.map(parseReview) || [],
+        pagination: {
+          total: response.pagination?.total || 0,
+          page: response.pagination?.page || 1,
+          perPage: response.pagination?.perPage || limit,
+          hasMore: response.pagination?.hasMore || false,
+        },
+      };
+    },
+
+    async getPublicServiceReviewById({ businessUnitId, reviewId }) {
+      if (!validateBusinessUnitId(businessUnitId)) {
+        throw new Error("Invalid business unit ID");
+      }
+      if (!validateReviewId(reviewId)) {
+        throw new Error("Invalid review ID");
+      }
+
+      const endpoint = buildUrl(ENDPOINTS.PUBLIC_REVIEW_BY_ID, { businessUnitId, reviewId });
+      const response = await this._makeRequest({ endpoint });
+      return parseReview(response);
+    },
+
+    // Private Service Review methods
     async getServiceReviews({
       businessUnitId,
       stars = null,
@@ -330,6 +386,66 @@ export default defineApp({
       return response;
     },
 
+    // Conversation methods
+    async getConversations({
+      limit = DEFAULT_LIMIT,
+      offset = 0,
+      sortBy = SORT_OPTIONS.CREATED_AT_DESC,
+      businessUnitId = null,
+    } = {}) {
+      const params = {
+        perPage: limit,
+        page: Math.floor(offset / limit) + 1,
+        orderBy: sortBy,
+      };
+
+      if (businessUnitId) {
+        params.businessUnitId = businessUnitId;
+      }
+
+      const response = await this._makeRequestWithRetry({
+        endpoint: ENDPOINTS.CONVERSATIONS,
+        params,
+      });
+
+      return {
+        conversations: response.conversations || [],
+        pagination: {
+          total: response.pagination?.total || 0,
+          page: response.pagination?.page || 1,
+          perPage: response.pagination?.perPage || limit,
+          hasMore: response.pagination?.hasMore || false,
+        },
+      };
+    },
+
+    async getConversationById({ conversationId }) {
+      if (!conversationId) {
+        throw new Error("Invalid conversation ID");
+      }
+
+      const endpoint = buildUrl(ENDPOINTS.CONVERSATION_BY_ID, { conversationId });
+      const response = await this._makeRequest({ endpoint });
+      return response;
+    },
+
+    async replyToConversation({ conversationId, message }) {
+      if (!conversationId) {
+        throw new Error("Invalid conversation ID");
+      }
+      if (!message || typeof message !== 'string') {
+        throw new Error("Reply message is required");
+      }
+
+      const endpoint = buildUrl(ENDPOINTS.REPLY_TO_CONVERSATION, { conversationId });
+      const response = await this._makeRequest({
+        endpoint,
+        method: "POST",
+        data: { message },
+      });
+      return response;
+    },
+
     // Webhook methods
     async createWebhook({ url, events = [], businessUnitId = null }) {
       if (!url) {

components/trustpilot/common/constants.mjs

@@ -16,6 +16,7 @@ export const ENDPOINTS = {
 
   // Public Reviews
   PUBLIC_REVIEWS: "/business-units/{businessUnitId}/reviews",
+  PUBLIC_REVIEW_BY_ID: "/business-units/{businessUnitId}/reviews/{reviewId}",
 
   // Private Reviews (Service)
   PRIVATE_SERVICE_REVIEWS: "/private/business-units/{businessUnitId}/reviews",
@@ -27,10 +28,15 @@ export const ENDPOINTS = {
   PRIVATE_PRODUCT_REVIEW_BY_ID: "/private/product-reviews/{reviewId}",
   REPLY_TO_PRODUCT_REVIEW: "/private/product-reviews/{reviewId}/reply",
 
+  // Conversations
+  CONVERSATIONS: "/private/conversations",
+  CONVERSATION_BY_ID: "/private/conversations/{conversationId}",
+  REPLY_TO_CONVERSATION: "/private/conversations/{conversationId}/reply",
+  
   // Invitations
   EMAIL_INVITATIONS: "/private/business-units/{businessUnitId}/email-invitations",
 
-  // Webhooks
+  // Webhooks (deprecated for polling)
   WEBHOOKS: "/private/webhooks",
   WEBHOOK_BY_ID: "/private/webhooks/{webhookId}",
 };
@@ -75,4 +81,18 @@ export const RETRY_CONFIG = {
   MAX_RETRIES: 3,
   INITIAL_DELAY: 1000,
   MAX_DELAY: 10000,
+};
+
+export const POLLING_CONFIG = {
+  DEFAULT_TIMER_INTERVAL_SECONDS: 15 * 60, // 15 minutes
+  MAX_ITEMS_PER_POLL: 100,
+  LOOKBACK_HOURS: 24, // How far back to look on first run
+};
+
+export const SOURCE_TYPES = {
+  NEW_REVIEWS: "new_reviews",
+  UPDATED_REVIEWS: "updated_reviews",
+  NEW_REPLIES: "new_replies", 
+  NEW_CONVERSATIONS: "new_conversations",
+  UPDATED_CONVERSATIONS: "updated_conversations",
 };

components/trustpilot/sources/common/polling.mjs

@@ -0,0 +1,170 @@
+import trustpilot from "../../app/trustpilot.app.ts";
+import { POLLING_CONFIG, SOURCE_TYPES } from "../../common/constants.mjs";
+
+export default {
+  props: {
+    trustpilot,
+    db: "$.service.db",
+    timer: {
+      type: "$.interface.timer",
+      default: {
+        intervalSeconds: POLLING_CONFIG.DEFAULT_TIMER_INTERVAL_SECONDS,
+      },
+    },
+    businessUnitId: {
+      propDefinition: [
+        trustpilot,
+        "businessUnitId",
+      ],
+      optional: true,
+      description: "Business Unit ID to filter events for. If not provided, will receive events for all business units.",
+    },
+  },
+  methods: {
+    _getLastPolled() {
+      return this.db.get("lastPolled");
+    },
+    _setLastPolled(timestamp) {
+      this.db.set("lastPolled", timestamp);
+    },
+    _getSeenItems() {
+      return this.db.get("seenItems") || {};
+    },
+    _setSeenItems(seenItems) {
+      this.db.set("seenItems", seenItems);
+    },
+    _cleanupSeenItems(seenItems, hoursToKeep = 72) {
+      const cutoff = Date.now() - (hoursToKeep * 60 * 60 * 1000);
+      const cleaned = {};
+      
+      Object.entries(seenItems).forEach(([key, timestamp]) => {
+        if (timestamp > cutoff) {
+          cleaned[key] = timestamp;
+        }
+      });
+      
+      return cleaned;
+    },
+    getSourceType() {
+      // Override in child classes
+      return SOURCE_TYPES.NEW_REVIEWS;
+    },
+    getPollingMethod() {
+      // Override in child classes to return the app method to call
+      throw new Error("getPollingMethod must be implemented in child class");
+    },
+    getPollingParams(since) {
+      // Override in child classes to return method-specific parameters
+      return {
+        businessUnitId: this.businessUnitId,
+        limit: POLLING_CONFIG.MAX_ITEMS_PER_POLL,
+        sortBy: "createdat.desc", // Most recent first
+      };
+    },
+    isNewItem(item, sourceType) {
+      // For "new" sources, check creation date
+      // For "updated" sources, check update date
+      const itemDate = sourceType.includes("updated") 
+        ? new Date(item.updatedAt) 
+        : new Date(item.createdAt || item.updatedAt);
+      
+      const lastPolled = this._getLastPolled();
+      return !lastPolled || itemDate > new Date(lastPolled);
+    },
+    generateDedupeKey(item, sourceType) {
+      // Create unique key: itemId + relevant timestamp
+      const timestamp = sourceType.includes("updated") 
+        ? item.updatedAt 
+        : (item.createdAt || item.updatedAt);
+      
+      return `${item.id}_${timestamp}`;
+    },
+    generateMeta(item, sourceType) {
+      const dedupeKey = this.generateDedupeKey(item, sourceType);
+      const summary = this.generateSummary(item, sourceType);
+      const timestamp = sourceType.includes("updated") 
+        ? item.updatedAt 
+        : (item.createdAt || item.updatedAt);
+      
+      return {
+        id: dedupeKey,
+        summary,
+        ts: new Date(timestamp).getTime(),
+      };
+    },
+    generateSummary(item, sourceType) {
+      // Override in child classes for specific summaries
+      return `${sourceType} - ${item.id}`;
+    },
+    async fetchItems(since) {
+      const method = this.getPollingMethod();
+      const params = this.getPollingParams(since);
+      
+      try {
+        const result = await this.trustpilot[method](params);
+        
+        // Handle different response formats
+        if (result.reviews) {
+          return result.reviews;
+        } else if (result.conversations) {
+          return result.conversations;
+        } else if (Array.isArray(result)) {
+          return result;
+        } else {
+          return [];
+        }
+      } catch (error) {
+        console.error(`Error fetching items with ${method}:`, error);
+        throw error;
+      }
+    },
+    async pollForItems() {
+      const sourceType = this.getSourceType();
+      const lastPolled = this._getLastPolled();
+      const seenItems = this._getSeenItems();
+      
+      // If first run, look back 24 hours
+      const since = lastPolled || new Date(Date.now() - (POLLING_CONFIG.LOOKBACK_HOURS * 60 * 60 * 1000)).toISOString();
+      
+      console.log(`Polling for ${sourceType} since ${since}`);
+      
+      try {
+        const items = await this.fetchItems(since);
+        const newItems = [];
+        const currentTime = Date.now();
+        
+        for (const item of items) {
+          // Check if item is new based on source type
+          if (this.isNewItem(item, sourceType)) {
+            const dedupeKey = this.generateDedupeKey(item, sourceType);
+            
+            // Check if we've already seen this exact item+timestamp
+            if (!seenItems[dedupeKey]) {
+              seenItems[dedupeKey] = currentTime;
+              newItems.push(item);
+            }
+          }
+        }
+        
+        // Emit new items
+        for (const item of newItems.reverse()) { // Oldest first
+          const meta = this.generateMeta(item, sourceType);
+          this.$emit(item, meta);
+        }
+        
+        // Update state
+        this._setLastPolled(new Date().toISOString());
+        this._setSeenItems(this._cleanupSeenItems(seenItems));
+        
+        console.log(`Found ${newItems.length} new items of type ${sourceType}`);
+        
+      } catch (error) {
+        console.error(`Polling failed for ${sourceType}:`, error);
+        throw error;
+      }
+    },
+  },
+  async run() {
+    await this.pollForItems();
+  },
+};