sources/new-scheduled-tasks/new-scheduled-tasks.mjs
import pipedream from "../../pipedream.app.mjs"; import sampleEmit from "./test-event.mjs"; import { uuid } from "uuidv4"; export default { key: "pipedream-new-scheduled-tasks", name: "New Scheduled Tasks", type: "source", description: "Exposes an HTTP API for scheduling messages to be emitted at a future time", version: "0.3.1", dedupe: "unique", props: { pipedream, secret: { type: "string", secret: true, label: "Secret", optional: true, description: "**Optional but recommended**: if you enter a secret here, you must pass this value in the `x-pd-secret` HTTP header when making requests", }, http: { label: "Endpoint", description: "The endpoint where you'll send task scheduler requests", type: "$.interface.http", customResponse: true, }, db: "$.service.db", }, methods: { selfChannel() { return "self"; }, queuedEventsChannel() { return "$in"; }, httpRespond({ status, body, }) { this.http.respond({ headers: { "content-type": "application/json", }, status, body, }); }, async selfSubscribe() { const isSubscribedToSelf = this.db.get("isSubscribedToSelf"); if (!isSubscribedToSelf) { const componentId = process.env.PD_COMPONENT; const selfChannel = this.selfChannel(); console.log(`Subscribing to ${selfChannel} channel for event source`); console.log( await this.pipedream.subscribe(componentId, componentId, selfChannel), ); this.db.set("isSubscribedToSelf", true); } }, validateEventBody(event, operation) { const errors = []; if (this.secret && event.headers["x-pd-secret"] !== this.secret) { errors.push( "Secret on incoming request doesn't match the configured secret", ); } if (operation === "schedule") { const { timestamp, message, } = event.body; const epoch = Date.parse(timestamp); if (!timestamp) { errors.push( "No timestamp included in payload. Please provide an ISO8601 timestamp in the 'timestamp' field", ); } if (timestamp && !epoch) { errors.push("Timestamp isn't a valid ISO 8601 string"); } if (!message) { errors.push("No message passed in payload"); } } return errors; }, scheduleTask(event) { const errors = this.validateEventBody(event, "schedule"); let status, body; if (errors.length) { console.log(errors); status = 400; body = { errors, }; } else { const id = this.emitScheduleEvent(event.body, event.body.timestamp); status = 200; body = { msg: "Successfully scheduled task", id, }; } this.httpRespond({ status, body, }); }, emitScheduleEvent(event, timestamp) { const selfChannel = this.selfChannel(); const epoch = Date.parse(timestamp); const $id = uuid(); console.log(`Scheduled event to emit on: ${new Date(epoch)}`); this.$emit( { ...event, $channel: selfChannel, $id, }, { name: selfChannel, id: $id, delivery_ts: epoch, }, ); return $id; }, async cancelTask(event) { const errors = this.validateEventBody(event, "cancel"); let status, msg; if (errors.length) { console.log(errors); status = 400; msg = "Secret on incoming request doesn't match the configured secret"; } else { try { const id = event.body.id; const isCanceled = await this.deleteEvent(event); if (isCanceled) { status = 200; msg = `Cancelled scheduled task for event ${id}`; } else { status = 404; msg = `No event with ${id} found`; } } catch (error) { console.log(error); status = 500; msg = "Failed to schedule task. Please see the logs"; } } this.httpRespond({ status, body: { msg, }, }); }, async deleteEvent(event) { const componentId = process.env.PD_COMPONENT; const inChannel = this.queuedEventsChannel(); const { id } = event.body; const events = await this.pipedream.listEvents( componentId, inChannel, ); console.log("Events: ", events); const eventToCancel = events.data.find((e) => { const { metadata } = e; return metadata.id === id; }); console.log("Event to cancel: ", eventToCancel); if (!eventToCancel) { console.log(`No event with ${id} found`); return false; } await this.pipedream.deleteEvent( componentId, eventToCancel.id, inChannel, ); return true; }, emitEvent(event, summary) { const id = event.$id; delete event.$channel; delete event.$id; this.$emit(event, { summary: summary ?? JSON.stringify(event), id, ts: +new Date(), }); }, }, async run(event) { await this.selfSubscribe(); const { path } = event; if (path === "/schedule") { this.scheduleTask(event); } else if (path === "/cancel") { await this.cancelTask(event); } else if (event.$channel === this.selfChannel()) { this.emitEvent(event); } }, sampleEmit, };
Chase Roberts@chsrbrts@benedictevans If you haven’t used @pipedream yet, then you haven’t lived.
✨Ellie Day✨@heyelliedayEvaluation update: @pipedream has quite literally been a dream to work with! I’m excited to leverage this tool for all the various workflows I need to write. I’m currently at 11k invocations a day from the initial workflows I’ve written in the past couple weeks.
Michael Braedley@MBraedleyUpdate: I got it working properly, and it's working so well that I'm dropping IFTTT. @pipedream can do everything that IFTTT basic can, and most (if not all things) IFTTT pro can for free or at a reasonable price if you need it. I am recommending it for basically any power user.
Thomas Cutting@mrthomascuttingWant quick+dirty integrations for a serverless workflow - @pipedream is my new go-to 😃
Matthew Roberts@mattdotrobertsday 013 - finally hit node js. This is the secret sauce of taking #nocode projects that one step further. Pumped about getting deeper into @pipedream now
Kenneth Auchenberg 💭@auchenbergYahoo Pipes is back! Kinda 😍 @pipedream
Raymond Camden 🥑@raymondcamdenAwesome video by the @pipedream folks showing real time twitter sentiment analysis integrated with Google Sheets. This is where Pipedream *really* shines, connecting systems together in easy workflows.
Nacho Caballero@nachocaballeroI couldn't recommend @pipedream more. It's an amazing service to integrate different APIs. Much more powerful than Zapier and more user-friendly than AWS Lambda. I'm very proud to wear this t-shirt #NoCode
Jason Snow@jyksnwDeveloped a working prototype environmental sensor IoT solution with @particle Photon, @pipedream, and @MongoDB with full graphing and alerting in less than a day! All amazing technology, will def. be exploring these more.
Steven Terrana@steven_terrana@burgwyn you've inspired me to finally set up my own blog. I'll make sure my first blog post explains the tech behind the setup. think @obsdmd + @GatsbyJS + @pipedream.
🚄 James Augeri, PhD@DotDotJamesWant to low-code your back end, need more horsepower than @Bubble / @KnackHQ, or just miss Yahoo! Pipes? Check out @PipeDream
Sébastien Chopin@AtinuxGitHub issues should be like @linear_app for maintainers. Looking forward more integrations with GH actions or tools like @pipedream 👀
Raul@raul_predescuIf you're a dev and not using @pipedream, you're missing out. Been using it for months, daily. FREE for devs. Plenty of integrations and good limits. Absolutely love it.
Bruno Skvorc@bitfallsSo @pipedream is pretty amazing. In 3 minutes I just made a flow which adds @rickastley's Never Gonna Give You Up to my @spotify playlist whenever a new pull request arrives in an old repo of mine.
Zach Lanich@ZachLanichUm, wow 🤯 @pipedream
Steven Bell@bellontechI just used @pipedream to build a Shopify App. Wow, they make small backed tasks easy.
Jay Hack 🎩🇺🇸@_jayhack_Very impressed with this bad boi - it reminds me of a @PalantirTech internal tool, but geared towards integrations instead of data analysis and far more customizable. Great expectations here 🚀🤩
Tree Sturgeon 🔥🚴♂️🌳@philsturgeonFor context this is day 2 of a really challenging and stupid migration from Notion to @airtable with disparate/missing data. It's going better than expected and thanks to @pipedream I don't have to bother the iOS dev to add W3W.