From 8cfff29d459b096adbab10b94721d1caa4d496a3 Mon Sep 17 00:00:00 2001 From: Lixiyuan Date: Tue, 23 Jun 2026 19:25:04 +0800 Subject: [PATCH 1/3] Add eTrusted new review source --- components/etrusted/package.json | 2 +- .../sources/new-review/new-review.mjs | 151 ++++++++++++++++++ .../sources/new-review/test-event.mjs | 16 ++ 3 files changed, 168 insertions(+), 1 deletion(-) create mode 100644 components/etrusted/sources/new-review/new-review.mjs create mode 100644 components/etrusted/sources/new-review/test-event.mjs diff --git a/components/etrusted/package.json b/components/etrusted/package.json index 1336934e1db82..640fda5bd5411 100644 --- a/components/etrusted/package.json +++ b/components/etrusted/package.json @@ -1,6 +1,6 @@ { "name": "@pipedream/etrusted", - "version": "0.2.0", + "version": "0.3.0", "description": "Pipedream eTrusted Components", "main": "etrusted.app.mjs", "keywords": [ diff --git a/components/etrusted/sources/new-review/new-review.mjs b/components/etrusted/sources/new-review/new-review.mjs new file mode 100644 index 0000000000000..c4c5fa09d030b --- /dev/null +++ b/components/etrusted/sources/new-review/new-review.mjs @@ -0,0 +1,151 @@ +import { DEFAULT_POLLING_SOURCE_TIMER_INTERVAL } from "@pipedream/platform"; +import { parseObject } from "../../common/utils.mjs"; +import etrusted from "../../etrusted.app.mjs"; +import sampleEmit from "./test-event.mjs"; + +export default { + key: "etrusted-new-review", + name: "New Review", + description: "Emit new event when a new review is submitted on an eTrusted channel. [See the documentation](https://developers.etrusted.com/reference/getreviews)", + version: "0.0.1", + type: "source", + dedupe: "unique", + props: { + etrusted, + db: "$.service.db", + timer: { + type: "$.interface.timer", + label: "Polling Schedule", + description: "How often to poll the eTrusted Reviews API.", + default: { + intervalSeconds: DEFAULT_POLLING_SOURCE_TIMER_INTERVAL, + }, + }, + channelId: { + propDefinition: [ + etrusted, + "channelId", + ], + type: "string[]", + optional: true, + }, + type: { + propDefinition: [ + etrusted, + "type", + ], + optional: true, + }, + status: { + propDefinition: [ + etrusted, + "status", + ], + optional: true, + }, + rating: { + propDefinition: [ + etrusted, + "rating", + ], + optional: true, + }, + ignoreStatements: { + propDefinition: [ + etrusted, + "ignoreStatements", + ], + optional: true, + }, + }, + methods: { + _getLastSubmittedAt() { + return this.db.get("lastSubmittedAt"); + }, + _setLastSubmittedAt(value) { + this.db.set("lastSubmittedAt", value); + }, + _getParams() { + return { + channels: parseObject(this.channelId)?.join(","), + type: parseObject(this.type)?.join(","), + status: parseObject(this.status)?.join(","), + rating: parseObject(this.rating)?.join(","), + ignoreStatements: this.ignoreStatements, + orderBy: "submittedAt", + }; + }, + _getSubmittedAt(review) { + return review.submittedAt ?? review.submitted_at ?? review.createdAt ?? review.created_at; + }, + _getTs(review) { + const submittedAt = this._getSubmittedAt(review); + return submittedAt + ? Date.parse(submittedAt) + : Date.now(); + }, + _getSummary(review) { + const rating = review.rating + ? ` (${review.rating} star${review.rating === 1 + ? "" + : "s"})` + : ""; + return `New review${rating}: ${review.title ?? review.id}`; + }, + _emitReview(review) { + this.$emit(review, { + id: review.id, + summary: this._getSummary(review), + ts: this._getTs(review), + }); + }, + async _getReviews({ + maxResults, + submittedAfter, + } = {}) { + const response = this.etrusted.paginate({ + fn: this.etrusted.getListOfReviews, + params: { + ...this._getParams(), + submittedAfter, + }, + maxResults, + }); + + const reviews = []; + for await (const review of response) { + reviews.push(review); + } + return reviews; + }, + _processReviews(reviews) { + if (!reviews.length) { + return; + } + + const newestSubmittedAt = this._getSubmittedAt(reviews[0]); + for (const review of reviews.reverse()) { + this._emitReview(review); + } + + if (newestSubmittedAt) { + this._setLastSubmittedAt(newestSubmittedAt); + } + }, + }, + hooks: { + async deploy() { + const reviews = await this._getReviews({ + maxResults: 10, + }); + this._processReviews(reviews); + }, + }, + async run() { + const reviews = await this._getReviews({ + submittedAfter: this._getLastSubmittedAt(), + }); + this._processReviews(reviews); + }, + sampleEmit, +}; diff --git a/components/etrusted/sources/new-review/test-event.mjs b/components/etrusted/sources/new-review/test-event.mjs new file mode 100644 index 0000000000000..b149b4e94f6a7 --- /dev/null +++ b/components/etrusted/sources/new-review/test-event.mjs @@ -0,0 +1,16 @@ +export default { + id: "rev-1234567890", + title: "Great service", + comment: "The order arrived quickly and support was helpful.", + rating: 5, + type: "SERVICE_REVIEW", + status: "APPROVED", + submittedAt: "2026-06-23T09:40:50.000Z", + channel: { + id: "chl-1234567890", + name: "Example channel", + }, + consumer: { + email: "customer@example.com", + }, +}; From 11e43e6e345ecc6e22d7006320fcddf916bef7b2 Mon Sep 17 00:00:00 2001 From: Lixiyuan Date: Tue, 23 Jun 2026 20:16:33 +0800 Subject: [PATCH 2/3] Handle eTrusted review cursor edge cases --- .../sources/new-review/new-review.mjs | 100 +++++++++++++++--- 1 file changed, 87 insertions(+), 13 deletions(-) diff --git a/components/etrusted/sources/new-review/new-review.mjs b/components/etrusted/sources/new-review/new-review.mjs index c4c5fa09d030b..6e9c0349e7efc 100644 --- a/components/etrusted/sources/new-review/new-review.mjs +++ b/components/etrusted/sources/new-review/new-review.mjs @@ -59,18 +59,30 @@ export default { }, }, methods: { - _getLastSubmittedAt() { - return this.db.get("lastSubmittedAt"); + _getCursor() { + return this.db.get("cursor") ?? { + submittedAt: this.db.get("lastSubmittedAt"), + ids: [], + }; + }, + _setCursor(cursor) { + this.db.set("cursor", cursor); }, - _setLastSubmittedAt(value) { - this.db.set("lastSubmittedAt", value); + _toCsv(value) { + const parsed = parseObject(value); + if (parsed == null || parsed === "") { + return undefined; + } + return Array.isArray(parsed) + ? parsed.join(",") + : String(parsed); }, _getParams() { return { - channels: parseObject(this.channelId)?.join(","), - type: parseObject(this.type)?.join(","), - status: parseObject(this.status)?.join(","), - rating: parseObject(this.rating)?.join(","), + channels: this._toCsv(this.channelId), + type: this._toCsv(this.type), + status: this._toCsv(this.status), + rating: this._toCsv(this.rating), ignoreStatements: this.ignoreStatements, orderBy: "submittedAt", }; @@ -99,6 +111,13 @@ export default { ts: this._getTs(review), }); }, + _getSubmittedAfter(cursor) { + if (!cursor?.submittedAt) { + return undefined; + } + + return new Date(Date.parse(cursor.submittedAt) - 1).toISOString(); + }, async _getReviews({ maxResults, submittedAfter, @@ -118,18 +137,72 @@ export default { } return reviews; }, + _isNewReview(review, cursor) { + if (!cursor?.submittedAt) { + return true; + } + + const submittedAt = this._getSubmittedAt(review); + if (!submittedAt) { + return true; + } + + const reviewTs = Date.parse(submittedAt); + const cursorTs = Date.parse(cursor.submittedAt); + if (reviewTs > cursorTs) { + return true; + } + if (reviewTs < cursorTs) { + return false; + } + + return !cursor.ids?.includes(review.id); + }, + _sortBySubmittedAt(reviews) { + return reviews.sort((a, b) => this._getTs(a) - this._getTs(b)); + }, + _getNextCursor(reviews, cursor) { + const withSubmittedAt = reviews.filter((review) => this._getSubmittedAt(review)); + if (!withSubmittedAt.length) { + return cursor; + } + + const newestSubmittedAt = withSubmittedAt.reduce((max, review) => { + const submittedAt = this._getSubmittedAt(review); + return !max || Date.parse(submittedAt) > Date.parse(max) + ? submittedAt + : max; + }, null); + const ids = withSubmittedAt + .filter((review) => this._getSubmittedAt(review) === newestSubmittedAt) + .map((review) => review.id); + + if (cursor?.submittedAt === newestSubmittedAt) { + ids.push(...(cursor.ids ?? [])); + } + + return { + submittedAt: newestSubmittedAt, + ids: [ + ...new Set(ids), + ], + }; + }, _processReviews(reviews) { if (!reviews.length) { return; } - const newestSubmittedAt = this._getSubmittedAt(reviews[0]); - for (const review of reviews.reverse()) { + const cursor = this._getCursor(); + const newReviews = reviews.filter((review) => this._isNewReview(review, cursor)); + const nextCursor = this._getNextCursor(reviews, cursor); + + for (const review of this._sortBySubmittedAt(newReviews)) { this._emitReview(review); } - if (newestSubmittedAt) { - this._setLastSubmittedAt(newestSubmittedAt); + if (nextCursor?.submittedAt) { + this._setCursor(nextCursor); } }, }, @@ -142,8 +215,9 @@ export default { }, }, async run() { + const cursor = this._getCursor(); const reviews = await this._getReviews({ - submittedAfter: this._getLastSubmittedAt(), + submittedAfter: this._getSubmittedAfter(cursor), }); this._processReviews(reviews); }, From 974a19a14fd8df6f4db1308181698aec89a291a7 Mon Sep 17 00:00:00 2001 From: Lixiyuan Date: Wed, 24 Jun 2026 00:06:30 +0800 Subject: [PATCH 3/3] Omit empty eTrusted review filters --- components/etrusted/sources/new-review/new-review.mjs | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/components/etrusted/sources/new-review/new-review.mjs b/components/etrusted/sources/new-review/new-review.mjs index 6e9c0349e7efc..3855414906f8f 100644 --- a/components/etrusted/sources/new-review/new-review.mjs +++ b/components/etrusted/sources/new-review/new-review.mjs @@ -73,9 +73,13 @@ export default { if (parsed == null || parsed === "") { return undefined; } - return Array.isArray(parsed) - ? parsed.join(",") - : String(parsed); + if (Array.isArray(parsed)) { + const values = parsed.filter((item) => item != null && item !== ""); + return values.length + ? values.join(",") + : undefined; + } + return String(parsed); }, _getParams() { return {