Community Watch
Category: Engineering & DevOps
This page is generated from the Air Pipe marketplace. Browse it live to install into your organization.
Watch a developer community for the things you care about. On a schedule it searches Hacker News for your keywords, has a model write a short digest, and sends it to Telegram — only when there is something worth saying.
The last part is the point. A monitor that messages you every six hours whether or not anything happened gets muted within a week.
The quiet-run gate
The model decides, and the config obeys it:
schema:
type: object
properties:
headline: { type: string }
worth_sending: { type: boolean }
bullets: { type: array, items: { type: string } }
required: [headline, worth_sending, bullets]
schema: is enforced by the provider, so worth_sending is always present and always a boolean —
there is no "the model forgot the field" branch to write. The send action is then gated on it:
- name: Send
run_on_assertion:
tests:
- action: Message
value: 0.json.send
is_equal_to: true
A quiet run makes no outbound call at all.
sent means delivered, not attempted. It is computed from Telegram's own HTTP status, and a
send that was attempted and did not land fails the whole run with 502 and
the digest was not delivered. This matters more than it looks: reporting the intent instead
means an expired bot token answers {"sent": true} with a 200 forever while nothing arrives —
a monitor that has stopped working and cannot tell you. All four paths are verified: quiet (200,
sent:false), delivered (200, sent:true), rejected 401 (502) and Telegram down 500 (502).
wanted_to_send is kept alongside it, so you can still see the model's call separately from
whether it reached anyone, and delivery_configured says whether Telegram was set up at all —
with no token the send is skipped rather than attempted, which is what keeps the
no-credentials-to-start run below honest instead of 502.
No credentials to start
Hacker News search needs no key, so you can run this end to end before you own anything:
curl -X POST https://<your-endpoint>/watch/run -H "x-api-key: $KEY"
# {"found": 20, "sent": true, "headline": "Mixed bag of developer projects and discussions"}
Add the OpenAI key for the digest and the Telegram pair for delivery.
Reddit works the same way — swap the search action's URL — but Reddit refuses anonymous requests from datacentre IPs (403), so it needs a registered OAuth app. Hacker News does not, which is why this pack ships against it.
Endpoints
| Method | Route | Purpose |
|---|---|---|
| POST | /watch/run | Search, summarise, and send if it is worth sending. |
Also runs on a schedule (0 */6 * * *). Call it by hand to see what it would send.
Setup
| Variable | Value |
|---|---|
WATCH_KEYWORDS | What to look for, e.g. workflow automation |
WATCH_WINDOW_HOURS | How far back each run looks. Match it to your cron — 6 on the default 6-hourly schedule. Defaults to 24 if unset. |
OPENAI_KEY | Writes the digest. |
TELEGRAM_BOT_TOKEN | From @BotFather, without the bot prefix. |
TELEGRAM_CHAT_ID | Your numeric id — message @userinfobot. |
Three things worth copying
Filtering is a transform, not a script. Picking and renaming fields is what jq is for, and it
costs no interpreter — measured at roughly 2ms per invocation saved against a code: action doing
the same reshape:
post_transforms:
- extract_with_jq: >
[ .body.hits[] | select(.title != null)
| { title: .title, url: .url, points: (.points // 0) } ]
The window is applied by the search, not after it. WATCH_WINDOW_HOURS becomes a
numericFilters=created_at_i>… on the query, so a run only ever receives recent items rather
than fetching the same top 20 every time and filtering locally.
This is a rolling window, not a watermark: set it wider than your cron and items are reported
twice; set it equal and each item is reported once. For true de-duplication across runs you want
a stored high-water mark — a state: action holding the last run's epoch, read at the start and
written at the end. That needs a durable state backend (AIRPIPE__DATABASE_URL); without one
state is per-process and in-memory, so a restart would silently re-send. The window needs nothing
at all, which is why it is the default here.
A code source is interpolated, but only halfway — on purpose. var/env/ap_var markers are
substituted by the config-level pass, which is why the delivery check below can read a variable
inline. Request-scoped markers are not: a|body::x| and a|SomeAction::y| are left standing
verbatim, because splicing request data into a program's text is injection — the same reason you
bind database parameters instead of concatenating them. Read another action through input: and
$input, or with $('Name'), and the value arrives as data:
- name: Outcome
input: a|Send| # resolved here, as data
code:
source: |
const status = Number($input.first()?.json?.status) || 0;
Getting this wrong is quiet rather than loud: the marker stays in the string, Number() yields
NaN, and the script computes something plausible and wrong.
A search that finds nothing is not an error. expect_status: [2xx, 404] — without it an empty
result fails the action, and the monitor breaks on the quiet day it was built for.
depends_on:, not run_when_succeeded:, after a gated action. Send is skipped on a quiet
run; Respond still has to answer. A success gate there would skip the response too, and the
caller would get nothing back.
Making it yours
- Another community: change
hn_apiand the jq inHits. Anything with a JSON search endpoint works — Lobsters, GitHub issues, a Discourse forum. - Somewhere else to send: replace the
Sendaction's URL and body. Slack and Discord webhooks are both a single POST. - A different bar for "interesting": it is one sentence in the
system:prompt.
Requires
- An Air Pipe engine with
agent: schema:(3.4.0+). - An OpenAI key, and a Telegram bot if you want delivery.
Configuration
config.yml
name: CommunityWatch
description: >
Watch a developer community for the things you care about. On a schedule it
searches Hacker News for your keywords, keeps what was posted inside a rolling
window, has a model write a short digest with a structured shape, and sends it
to Telegram — but only when there is something to say.
The window is WATCH_WINDOW_HOURS wide and is applied by the search itself, so a
run only ever sees recent items. Set it to match your cron: on the default
6-hourly schedule a 6-hour window means each item is reported once. It is a
window, not a watermark -- see the README for durable de-duplication.
docs: true
# Managed variables:
# WATCH_KEYWORDS — comma-separated, e.g. "airpipe,n8n,workflow automation"
# WATCH_WINDOW_HOURS — how far back to look, e.g. 24
# OPENAI_KEY — writes the digest
# TELEGRAM_BOT_TOKEN — from @BotFather, without the "bot" prefix
# TELEGRAM_CHAT_ID — where the digest goes
global:
variables:
hn_api: "https://hn.algolia.com/api/v1/search_by_date"
openai: "https://api.openai.com/v1"
telegram_api: "https://api.telegram.org"
interfaces:
watch/run:
output: http
method: POST
summary: Search a community for your keywords and send a digest
description: >
Runs on a schedule, or call it by hand to see what it would send.
Returns the digest it produced and whether it sent anything.
tags: [monitoring, hacker-news, openai, telegram, digest]
response_example: |
{"found": 4, "sent": true, "headline": "Four threads on workflow automation"}
schedule:
enabled: true
cron: "0 */6 * * *"
actions:
# ── what are we looking for, and how far back ────────────────────────
# One action holds the search window so the rest of the config never does
# date maths. `->epoch` turns a timestamp into seconds, which is what the
# Hacker News API wants for a numeric filter.
# WATCH_WINDOW_HOURS is text by the time this runs -- interpolation happens
# before the script does -- and an unset variable renders as the string
# "null", which is why the fallback is a Number() check rather than `??`.
- name: Window
input: a|body|
code:
language: js
source: |
const raw = Number("a|ap_var::WATCH_WINDOW_HOURS|");
const hours = Number.isFinite(raw) && raw > 0 ? raw : 24;
return [{ json: {
hours,
since: Math.floor(Date.now() / 1000) - Math.round(hours * 3600),
} }];
# ── search ──────────────────────────────────────────────────────────
# Algolia's HN index needs no credentials, which is why this pack can be
# run end to end before you own anything.
- name: Search
run_when_succeeded: [Window]
http:
method: GET
url: "a|var::hn_api|?query=a|ap_var::WATCH_KEYWORDS|&tags=(story,comment)&hitsPerPage=20&numericFilters=created_at_i>a|Window::0.json.since|"
# A community search legitimately returns nothing, and an empty result
# is information rather than a failure.
expect_status: [2xx, 404]
# ── keep what is recent and worth reading ───────────────────────────
# A transform rather than a code: action -- picking and filtering fields is
# what jq is for, and it costs no interpreter.
- name: Hits
depends_on: [Search]
input: a|Search|
post_transforms:
- extract_with_jq: >
[ .body.hits[]
| select(.title != null or .comment_text != null)
| { title: (.title // (.story_title + " (comment)")),
url: (.url // ("https://news.ycombinator.com/item?id=" + .objectID)),
author: .author,
points: (.points // 0),
comments: (.num_comments // 0),
at: .created_at } ]
# ── write the digest, with a shape we can rely on ───────────────────
# `schema:` is enforced by the provider, so `content` comes back parsed and
# no action downstream has to unpack a string or guess at a field.
- name: Digest
depends_on: [Hits]
agent:
url: "a|var::openai|/chat/completions"
model: gpt-4o-mini
headers:
Authorization: "Bearer a|ap_var::OPENAI_KEY|"
system: |
You summarise developer-community activity for a busy reader.
Write a one-line headline, then at most five bullets, each naming the
thread and why it matters. Be plain: no hype, no filler. If the items
are not genuinely interesting, say so in the headline and keep the
bullets empty.
# A STRING, not the array itself: an array `input:` is read as a chat
# transcript of {role, content} objects, so passing data that way fails with
# "Missing required parameter: messages[1].role". Wrap it in text instead.
input: |
Here is what the search returned. Summarise it.
a|Hits|
schema:
type: object
properties:
headline: { type: string }
worth_sending: { type: boolean }
bullets:
type: array
items: { type: string }
required: [headline, worth_sending, bullets]
additionalProperties: false
# ── send, only when there is something to say ───────────────────────
- name: Message
depends_on: [Digest]
input: a|Digest|
code:
language: js
source: |
// JSON-escaped here rather than quoted in the request body: a headline
// containing a quote or a newline would otherwise break the body.
const hits = $('Hits').all() ?? []; // .all() is every item; .first() is item ONE
const d = $('Digest').first().json.content;
const lines = [d.headline, '', ...(d.bullets || []).map(b => '• ' + b)];
const text = lines.join('\n').trim();
// Is delivery configured? Without this check the pack cannot do its own
// headline trick -- run end to end before you own anything -- because an
// absent token would attempt a send, fail, and 502.
//
// These two markers are substituted by the CONFIG-level pass, which is the
// only interpolation a code source gets: it resolves var/env/ap_var roots
// and nothing else. Request-scoped markers -- the request body, or another
// action's output -- are deliberately NOT substituted here, because splicing
// request data into a program's text is injection. Read those through input:.
//
// An unset variable looks DIFFERENT depending on where it is read: in a
// json_output it renders as the text "null"; in a code source the marker
// is left standing verbatim. Both mean "not set", so both are tested.
const isSet = v => !!v && v !== "null" && !v.startsWith("a|");
const configured = isSet("a|ap_var::TELEGRAM_BOT_TOKEN|")
&& isSet("a|ap_var::TELEGRAM_CHAT_ID|");
return [{ json: { text, text_json: JSON.stringify(text),
worth_sending: !!d.worth_sending,
configured,
send: !!d.worth_sending && configured,
found: hits.length } }];
- name: Send
depends_on: [Message]
run_on_assertion:
tests:
- action: Message
value: 0.json.send
is_equal_to: true
http:
method: POST
url: "a|var::telegram_api|/bota|ap_var::TELEGRAM_BOT_TOKEN|/sendMessage"
headers:
content-type: application/json
body: |
{
"chat_id": a|ap_var::TELEGRAM_CHAT_ID|,
"text": a|Message::0.json.text_json|,
"disable_web_page_preview": true
}
# depends_on, not run_when_succeeded: a quiet run skips Send, and this must
# still answer rather than being skipped along with it -- a skipped action
# propagates the skip to anything gated on its success.
#
# `sent` reads Send's OWN status rather than Message's intent. It used to
# report the intent, so a rejected token (Telegram 401) still answered
# "sent": true and the run stayed green: a digest bot that had stopped
# delivering looked identical to one that was working. A skipped Send has no
# status and renders as the text "null", which is neither "200" nor a
# failure, so the quiet case stays honest too.
# Turns Send's outcome into the booleans the response promises.
- name: Outcome
depends_on: [Send]
input: a|Send|
code:
language: js
source: |
// Read Send through `input:`, not through an action-reference marker. Those
// do NOT resolve inside a code source (only var/ap_var do), so the marker
// form scored every delivery 0 and failed a successful send with 502.
const sent = $input.first()?.json ?? {};
const status = Number(sent.status) || 0;
return [{ json: { status, sent: status === 200 } }];
- name: Respond
run_when_succeeded: [Outcome]
json_output: |-
{
"found": a|Message::0.json.found|,
"wanted_to_send": a|Message::0.json.worth_sending|,
"delivery_configured": a|Message::0.json.configured|,
"sent": a|Outcome::0.json.sent|,
"headline": "a|Digest::content.headline|"
}
# A send that was attempted and did not land must fail the run. Without this
# the interface answers 200 whatever Telegram said, so a scheduled monitor
# reports success forever while delivering nothing -- the one failure mode a
# monitor must not have. Gated on the intent, so a quiet run never reaches it.
- name: Delivered
depends_on: [Respond]
run_on_assertion:
tests:
- action: Message
value: 0.json.send
is_equal_to: true
json_output: |-
{ "status": a|Outcome::0.json.status| }
assert:
error_message: "the digest was not delivered — check TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID"
http_code_on_error: 502
tests:
- value: status
is_equal_to: 200