more comments

Author: Afstkla

components/trustpilot/sources/common/polling.mjs

@@ -1,4 +1,5 @@
 import trustpilot from "../../trustpilot.app.mjs";
+import { DEFAULT_LIMIT } from "../../common/constants.mjs";
 
 /**
  * Base polling source for Trustpilot integration
@@ -25,10 +26,10 @@ export default {
   },
   methods: {
     _getLastReviewTime() {
-      return this.db.get("lastReviewTime");
+      return this.db.get(`lastReviewTime:${this.businessUnitId}`);
     },
     _setLastReviewTime(time) {
-      this.db.set("lastReviewTime", time);
+      this.db.set(`lastReviewTime:${this.businessUnitId}`, time);
     },
     /**
      * Override in child classes to provide review type-specific summary
@@ -40,10 +41,13 @@ export default {
       throw new Error("generateSummary must be implemented in child class");
     },
     /**
-     * Override in child classes to fetch reviews using appropriate method
+     * Override in child classes to fetch reviews.
+     * Requirements:
+     *   - Must return ALL reviews newer than `lastReviewTime` (handle pagination internally), or
+     *   - Return the first page AND expose a pagination cursor so the base can iterate (future).
      * @param {Object} _$ - Pipedream step context
-     * @param {Object} _params - Fetch parameters
-     * @returns {Object} - API response with reviews array
+     * @param {Object} _params - Fetch parameters produced by `getFetchParams(lastReviewTime)`
+     * @returns {{ reviews: Array }} - Array of normalized reviews
      */
     // eslint-disable-next-line no-unused-vars
     async fetchReviews(_$, _params) {
@@ -58,7 +62,7 @@ export default {
     getFetchParams(_lastReviewTime) {
       return {
         businessUnitId: this.businessUnitId,
-        perPage: 100,
+        perPage: DEFAULT_LIMIT,
       };
     },
     /**
@@ -69,8 +73,14 @@ export default {
      */
     // eslint-disable-next-line no-unused-vars
     filterNewReviews(reviews, _lastReviewTime) {
-      // Default: return all reviews (for APIs with server-side time filtering)
-      return reviews;
+      if (!_lastReviewTime) return reviews;
+      const lastMs = Date.parse(_lastReviewTime);
+      if (Number.isNaN(lastMs)) return reviews;
+
+      return reviews.filter((r) => {
+        const ms = Date.parse(r?.createdAt);
+        return Number.isFinite(ms) && ms > lastMs;
+      });
     },
   },
   async run({ $ }) {
@@ -102,16 +112,24 @@ export default {
 
     for (const review of newReviews) {
       // Track the latest review time
-      const reviewTime = new Date(review.createdAt).toISOString();
-      if (!latestReviewTime || new Date(reviewTime) > new Date(latestReviewTime)) {
+      const createdMs = Date.parse(review?.createdAt);
+      if (!Number.isFinite(createdMs)) {
+        ($.logger?.warn ?? console.warn)("Skipping review with invalid createdAt", {
+          id: review?.id,
+          createdAt: review?.createdAt,
+        });
+        continue;
+      }
+      const reviewTime = new Date(createdMs).toISOString();
+      if (!latestReviewTime || createdMs > Date.parse(latestReviewTime)) {
         latestReviewTime = reviewTime;
       }
 
       // Emit the review with unique ID and summary
       this.$emit(review, {
         id: review.id,
         summary: this.generateSummary(review),
-        ts: new Date(review.createdAt).getTime(),
+        ts: createdMs,
       });
     }
 

components/trustpilot/sources/new-product-reviews/new-product-reviews.mjs

@@ -1,5 +1,8 @@
 import common from "../common/polling.mjs";
-import { DEFAULT_LIMIT } from "../../common/constants.mjs";
+import {
+  DEFAULT_LIMIT,
+  MAX_LIMIT,
+} from "../../common/constants.mjs";
 
 export default {
   ...common,
@@ -29,8 +32,39 @@ export default {
       };
     },
     async fetchReviews($, params) {
-      // Use the shared method from the app directly
-      return await this.trustpilot.fetchProductReviews($, params);
+      const perPage = params.perPage ?? DEFAULT_LIMIT;
+      let page = params.page ?? 1;
+
+      // fetch first page
+      let result = await this.trustpilot.fetchProductReviews($, {
+        ...params,
+        page,
+      });
+      let all = Array.isArray(result.reviews)
+        ? result.reviews
+        : [];
+      let lastPageSize = all.length;
+
+      // keep paging while we get a full page and stay under MAX_LIMIT
+      while (lastPageSize === perPage && all.length < MAX_LIMIT) {
+        page += 1;
+        const next = await this.trustpilot.fetchProductReviews($, {
+          ...params,
+          page,
+        });
+        const chunk = Array.isArray(next.reviews) ?
+          next.reviews :
+          [];
+        if (chunk.length === 0) break;
+
+        all = all.concat(chunk);
+        lastPageSize = chunk.length;
+        result = next; // preserve any metadata from the latest fetch
+      }
+
+      // truncate to MAX_LIMIT in case there are more than allowed
+      result.reviews = all.slice(0, MAX_LIMIT);
+      return result;
     },
     filterNewReviews(reviews, lastReviewTime) {
       // Product reviews require client-side filtering since API doesn't support

components/trustpilot/sources/new-service-reviews/new-service-reviews.mjs

@@ -8,7 +8,7 @@ export default {
   ...common,
   key: "trustpilot-new-service-reviews",
   name: "New Service Reviews",
-  description: "Emit new event when a customer posts a new service review on Trustpilot. This source periodically polls the Trustpilot API to detect new service reviews using the private reviews API for comprehensive coverage. Each event contains the complete review data including star rating, review text, consumer details, business unit info, customer email, and timestamps. Ideal for monitoring overall business reputation, tracking customer satisfaction metrics, and triggering workflows based on review ratings or content.",
+  description: "Emit new event when a customer posts a new service review on Trustpilot. This source periodically polls the Trustpilot API to detect new service reviews using the private reviews API for comprehensive coverage.",
   version: "0.1.0",
   type: "source",
   dedupe: "unique",
@@ -46,8 +46,9 @@ export default {
           const nextResult = await this.trustpilot.fetchServiceReviews($, params);
           result.reviews = result.reviews.concat(nextResult.reviews || []);
 
-          if ((nextResult.reviews && nextResult.reviews.length < DEFAULT_LIMIT)
-            || result.reviews.length >= MAX_LIMIT) {
+          if (!nextResult.reviews ||
+            nextResult.reviews.length < DEFAULT_LIMIT ||
+            result.reviews.length >= MAX_LIMIT) {
             break;
           }
         }

components/trustpilot/trustpilot.app.mjs

@@ -59,12 +59,6 @@ export default {
       description: "Filter by SKU",
       optional: true,
     },
-    productUrl: {
-      type: "string",
-      label: "Product URL",
-      description: "Filter by product URL",
-      optional: true,
-    },
     page: {
       type: "integer",
       label: "Page",
@@ -162,10 +156,16 @@ export default {
       // Handle the correct response structure (reviews array)
       const reviews = response.reviews?.map(parseServiceReview) || [];
       const pagination = {
-        total: response.pagination?.total || 0,
-        page: response.pagination?.page || queryParams.page,
-        perPage: response.pagination?.perPage || queryParams.perPage,
-        hasMore: response.pagination?.hasMore || false,
+        total: typeof response.total === "number"
+          ? response.total :
+          null,
+        // Preserve the page and perPage we requested
+        page: queryParams.page,
+        perPage: queryParams.perPage,
+        // Determine if there’s a next page by checking for a "next" link
+        hasMore: Array.isArray(response.links)
+          ? response.links.some((l) => l?.rel === "next-page")
+          : false,
       };
 
       return {
@@ -195,6 +195,9 @@ export default {
       if (!businessUnitId) {
         throw new Error("Business Unit ID is required");
       }
+      if (!validateBusinessUnitId(businessUnitId)) {
+        throw new Error("Invalid business unit ID format");
+      }
 
       // Build the endpoint URL
       const endpoint = buildUrl(ENDPOINTS.PRIVATE_PRODUCT_REVIEWS, {
@@ -208,7 +211,6 @@ export default {
         locale,
         perPage,
         page,
-        includeReportedReviews: false,
         language,
       };