# CountWatch recipe: count in your app

Counts events where they happen (orders placed, e-mails sent, jobs done, logins failed) with a small counter in the process. Once an hour it sends every finished 5-minute bucket, **zeros included**, as one request: the watch is an explicit source, so if the app stops sending, the no-data alert fires. A failed send keeps its buckets for the next one (up to 6 hours). Set `COUNTWATCH_API_KEY` (and `COUNTWATCH_API_URL=https://staging.countwatch.com`) in the app's environment.

Count outcomes, not attempts: `count("orders")` after the order is saved, not when checkout starts. The calls are those of the client libraries (`count(metric, n, dimensions)`), so moving to `npm install countwatch` or `pip install countwatch` later changes only the import.

## Node (18+, no packages)

Save as `countwatch-counter.mjs`:

```js
// countwatch-counter.mjs — CountWatch counter (Node 18+, no packages). Same calls as the countwatch npm package.
// const cw = countwatch(); cw.count("orders", 1, { shop: "eu" }); … on shutdown: await cw.flush(); cw.stop();
const BUCKET = 300_000;
const MAX_BACK = 6 * 3_600_000 - BUCKET;

export function countwatch({ token = process.env.COUNTWATCH_API_KEY, api = process.env.COUNTWATCH_API_URL ?? "https://api.countwatch.com", everyMs = 3_600_000 } = {}) {
  const watches = new Map(); // key -> { metric, dimensions, from, counts: Map<bucket start, n> }
  const bucketOf = (t) => Math.floor(t / BUCKET) * BUCKET;

  function count(metric, n = 1, dimensions = {}) {
    const key = metric + "|" + JSON.stringify(Object.entries(dimensions).sort());
    const b = bucketOf(Date.now());
    let w = watches.get(key);
    if (!w) watches.set(key, (w = { metric, dimensions, from: b, counts: new Map() }));
    w.counts.set(b, (w.counts.get(b) ?? 0) + n);
  }

  let inFlight = null; // one flush at a time: a second call waits for the running one
  function flush() {
    inFlight ??= send().finally(() => (inFlight = null));
    return inFlight;
  }

  async function send() {
    const open = bucketOf(Date.now()); // the bucket still filling is sent next time
    const points = [];
    for (const w of watches.values()) {
      w.from = Math.max(w.from, open - MAX_BACK); // older buckets are no longer accepted: forget them
      for (const t of w.counts.keys()) if (t < w.from) w.counts.delete(t);
      for (let t = w.from; t < open; t += BUCKET) {
        points.push({ metric: w.metric, dimensions: w.dimensions, ts: new Date(t).toISOString(), set: w.counts.get(t) ?? 0 });
      }
    }
    for (let i = 0; i < points.length; i += 10_000) {
      const res = await fetch(api + "/v1/ingest", {
        method: "POST",
        headers: { authorization: "Bearer " + token, "content-type": "application/json" },
        body: JSON.stringify({ points: points.slice(i, i + 10_000) }),
        signal: AbortSignal.timeout(10_000),
      });
      if (!res.ok) throw new Error("CountWatch " + res.status + ": " + (await res.text()).slice(0, 200));
      const answer = await res.json().catch(() => ({}));
      // Refused points (a watch limit, a bad name) are not retried: say so.
      if (answer.rejected?.length) console.error("countwatch: points refused:", JSON.stringify(answer.rejected).slice(0, 300));
    }
    for (const w of watches.values()) {
      w.from = Math.max(w.from, open);
      for (const t of w.counts.keys()) if (t < open) w.counts.delete(t);
    }
  }

  // The first send waits a random part of the interval, so processes started together (a deploy) send at different times.
  const tick = () => flush().catch((e) => console.error("countwatch:", e.message));
  let timer = setTimeout(() => {
    tick();
    timer = setInterval(tick, everyMs);
    timer.unref?.();
  }, Math.random() * everyMs);
  timer.unref?.();
  return { count, flush, stop: () => clearTimeout(timer) }; // clears the first timeout or the interval
}
```

```js
import { countwatch } from "./countwatch-counter.mjs";
const cw = countwatch();
cw.count("orders", 0, { shop: "eu" }); // declare each watch at startup: zeros go out even before the first event
// … where the event happens:
cw.count("orders", 1, { shop: "eu" });
// on shutdown:
process.on("SIGTERM", async () => { await cw.flush().catch(() => {}); cw.stop(); process.exit(0); });
```

## Python (3.8+, standard library)

Save as `countwatch_counter.py` (not `countwatch.py`: that name would hide the `countwatch` package):

```python
# countwatch_counter.py — CountWatch counter (Python 3.8+, standard library only). Same calls as the countwatch
# PyPI package. cw = CountWatch(); cw.count("orders", 1, {"shop": "eu"}); … on shutdown: cw.flush(); cw.stop()
import json, os, random, threading, time, urllib.request
from datetime import datetime, timezone

BUCKET = 300
MAX_BACK = 6 * 3600 - BUCKET


class CountWatch:
    def __init__(self, token=None, api=None, every_s=3600):
        self.token = token or os.environ["COUNTWATCH_API_KEY"]
        self.api = api or os.environ.get("COUNTWATCH_API_URL", "https://api.countwatch.com")
        self.every_s = every_s
        self.watches = {}  # key -> {"metric", "dimensions", "from", "counts": {bucket start: n}}
        self.lock = threading.Lock()
        self.flushing = threading.Lock()  # one flush at a time
        self.timer = None
        self.stopped = False
        # The first send waits a random part of the interval, so processes started together (a deploy) send at different times.
        self._schedule(random.uniform(0, every_s))

    def count(self, metric, n=1, dimensions=None):
        dimensions = dimensions or {}
        key = metric + "|" + json.dumps(sorted(dimensions.items()))
        b = int(time.time()) // BUCKET * BUCKET
        with self.lock:
            w = self.watches.setdefault(key, {"metric": metric, "dimensions": dimensions, "from": b, "counts": {}})
            w["counts"][b] = w["counts"].get(b, 0) + n

    def flush(self):
        with self.flushing:
            self._send()

    def _send(self):
        open_b = int(time.time()) // BUCKET * BUCKET  # the bucket still filling is sent next time
        with self.lock:
            points = []
            for w in self.watches.values():
                w["from"] = max(w["from"], open_b - MAX_BACK)  # older buckets are no longer accepted: forget them
                w["counts"] = {t: n for t, n in w["counts"].items() if t >= w["from"]}
                for t in range(w["from"], open_b, BUCKET):
                    ts = datetime.fromtimestamp(t, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
                    points.append({"metric": w["metric"], "dimensions": w["dimensions"], "ts": ts, "set": w["counts"].get(t, 0)})
        for i in range(0, len(points), 10000):
            req = urllib.request.Request(
                self.api + "/v1/ingest",
                data=json.dumps({"points": points[i : i + 10000]}).encode(),
                headers={"authorization": "Bearer " + self.token, "content-type": "application/json"},
                method="POST",
            )
            answer = json.loads(urllib.request.urlopen(req, timeout=10).read() or b"{}")  # raises on a non-2xx answer: the buckets stay
            if answer.get("rejected"):  # refused points (a watch limit, a bad name) are not retried: say so
                print("countwatch: points refused:", json.dumps(answer["rejected"])[:300])
        with self.lock:
            for w in self.watches.values():
                w["from"] = max(w["from"], open_b)
                w["counts"] = {t: n for t, n in w["counts"].items() if t >= open_b}

    def _schedule(self, delay=None):
        if self.stopped:
            return
        self.timer = threading.Timer(self.every_s if delay is None else delay, self._tick)
        self.timer.daemon = True
        self.timer.start()

    def _tick(self):
        try:
            self.flush()
        except Exception as e:  # noqa: BLE001 — a failed send is retried next time
            print("countwatch:", e)
        self._schedule()

    def stop(self):
        self.stopped = True
        if self.timer:
            self.timer.cancel()
```

```python
from countwatch_counter import CountWatch
import atexit
cw = CountWatch()
cw.count("orders", 0, {"shop": "eu"})  # declare each watch at startup: zeros go out even before the first event
atexit.register(cw.flush)  # send the finished buckets on a normal exit
# … where the event happens:
cw.count("orders", 1, {"shop": "eu"})
```

## Several processes

Each point `set`s its bucket, so two processes sending the same watch overwrite each other. Give each process its own dimension (`{"instance": "web-1"}`; mind the plan's watch limit), or count in one place.

## History

Count the same events per 5 minutes from your database or logs for the last 89 days, every bucket (an empty one is a `0`), with UTC bucket starts, and send them once with `POST /v1/import` (`resolution: "5m"`, the same metric and dimensions). The API stops an import at a watch's first live bucket, so it never overwrites the counter's sends. Use `"1h"` or `"1d"` only when the source keeps nothing finer: coarser counts are spread evenly over their 5-minute buckets.

## Check it works

1. **Data arrives.** `check_setup` (or `curl -s https://staging.countwatch.com/v1/status -H "Authorization: Bearer $COUNTWATCH_API_KEY"`): the `orders` watch shows `last_request_at`, and `missing_buckets_24h` stays `0` (explicit watches). Do what `next_actions` says.
2. **History is in.** `GET /v1/imports/{import_id}` shows `status: done` and the rules chosen for `orders`, with the alerts they would have raised. (Skipped the import? The watch learns for about 3 weeks; starter rules apply meanwhile.)
3. **An alert reaches you.** The person turns on e-mail alerts in the dashboard (https://staging.countwatch.com/dashboard), or you add a webhook (`manage add_destination`, then `verify_destination`; the endpoint answers the challenge, see https://staging.countwatch.com/agent). Then `manage {"action": "send_test_alert"}`: every active destination receives a test alert (`test: true`) within a few minutes.