Build a Modern Data Stack for SaaS Analytics

brew tap airbytehq/tap && brew install abctl   # or unpack the release archive and put abctl on your PATH
abctl local install                            # UI at http://localhost:8000
abctl local credentials                        # prints the generated admin password

# pipelines/slack_activity.py
import dlt
from dlt.sources.helpers import requests

CHANNELS = ["C0GENERAL", "C0PRODUCT", "C0ENGINEERING"]


@dlt.source(name="slack_activity")
def slack_activity(slack_token: str = dlt.secrets.value):
    @dlt.resource(write_disposition="merge", primary_key=["channel", "ts"])
    def messages(
        last_ts=dlt.sources.incremental("ts", initial_value="0"),
    ):
        headers = {"Authorization": f"Bearer {slack_token}"}
        for channel in CHANNELS:
            cursor = None
            while True:
                params = {"channel": channel, "oldest": last_ts.last_value, "limit": 200}
                if cursor:
                    params["cursor"] = cursor
                resp = requests.get(
                    "https://slack.com/api/conversations.history",
                    params=params,
                    headers=headers,
                )
                data = resp.json()
                if not data.get("ok"):
                    raise RuntimeError(f"Slack error for {channel}: {data.get('error')}")
                for msg in data["messages"]:
                    yield {
                        "ts": msg["ts"],
                        "channel": channel,
                        "user": msg.get("user"),
                        "text_length": len(msg.get("text", "")),
                        "reaction_count": len(msg.get("reactions", [])),
                        "thread_reply_count": msg.get("reply_count", 0),
                        "has_attachment": bool(msg.get("files")),
                    }
                cursor = data.get("response_metadata", {}).get("next_cursor")
                if not cursor:
                    break

    return messages


if __name__ == "__main__":
    pipeline = dlt.pipeline(
        pipeline_name="slack_activity",
        destination="bigquery",
        dataset_name="raw_slack",
    )
    print(pipeline.run(slack_activity()))

-- models/staging/stripe/stg_stripe__subscription_mrr.sql
-- One row per subscription per month, from paid invoices on monthly plans.
-- (Annual plans need amount_paid / 12 spread over 12 months; add that when you have them.)
select
    subscription as subscription_id,
    date_trunc(date(created), month) as month,
    sum(amount_paid) / 100 as mrr
from {{ source('raw_stripe', 'invoices') }}
where subscription is not null
  and status = 'paid'
group by 1, 2

-- models/marts/finance/mrr_events.sql
-- Fill every month between a subscription's first invoice and today with 0,
-- so a cancellation shows up as a month with MRR 0.
with bounds as (
    select subscription_id, min(month) as first_month
    from {{ ref('stg_stripe__subscription_mrr') }}
    group by 1
),
spine as (
    select subscription_id, month
    from bounds,
        unnest(generate_date_array(first_month, date_trunc(current_date(), month), interval 1 month)) as month
),
monthly as (
    select
        spine.subscription_id,
        spine.month,
        coalesce(s.mrr, 0) as mrr
    from spine
    left join {{ ref('stg_stripe__subscription_mrr') }} s
        on s.subscription_id = spine.subscription_id and s.month = spine.month
),
changes as (
    select
        subscription_id,
        month as event_date,
        mrr - coalesce(lag(mrr) over (partition by subscription_id order by month), 0) as mrr_change,
        lag(mrr) over (partition by subscription_id order by month) as previous_mrr,
        mrr
    from monthly
)
select
    *,
    case
        when previous_mrr is null then 'new'
        when mrr = 0 then 'churn'
        when previous_mrr = 0 then 'reactivation'
        when mrr > previous_mrr then 'expansion'
        else 'contraction'
    end as event_type
from changes
where mrr_change != 0

pip install soda-core soda-bigquery

# soda/ds_config.yml
type: bigquery
name: warehouse
connection:
  account_info_json_path: ${env.GOOGLE_APPLICATION_CREDENTIALS}
  dataset: raw_stripe
  project_id: ${env.BQ_PROJECT_ID}

# soda/contracts/stripe_invoices.yml
# dataset format: <data source>/<project>/<dataset>/<table>
dataset: warehouse/my-project/raw_stripe/invoices
columns:
  - name: amount_paid
    checks:
      - missing:
      - invalid:
          valid_min: 0
          name: No negative invoice amounts
  - name: currency
    checks:
      - invalid:
          valid_values: [usd, eur, gbp]
          name: Only expected currencies
checks:
  - row_count:
  - freshness:
      column: created
      threshold:
        unit: hour
        must_be_less_than: 3

soda contract test --contract soda/contracts/stripe_invoices.yml     # syntax only, no data read
soda contract verify --contract soda/contracts/stripe_invoices.yml -ds soda/ds_config.yml

# Summary:
|----------------|---|----|
| Checks         | 5 |    |
| Passed         | 5 | ✅ |
| Failed         | 0 | ✅ |
| Warned         | 0 | ✅ |
| Not Evaluated  | 0 | ✅ |
| Excluded       | 0 | ✅ |
| Runtime Errors | 0 | ✅ |

docker run -d --name metabase -p 127.0.0.1:3000:3000 \
  -e MB_DB_TYPE=postgres -e MB_DB_HOST=db.internal -e MB_DB_PORT=5432 \
  -e MB_DB_DBNAME=metabase -e MB_DB_USER=metabase -e MB_DB_PASS="$METABASE_DB_PASSWORD" \
  metabase/metabase:latest

-- Metabase question: MRR Waterfall
WITH monthly AS (
  SELECT
    DATE_TRUNC(event_date, MONTH) AS month,
    SUM(IF(event_type = 'new', mrr_change, 0)) AS new_mrr,
    SUM(IF(event_type = 'expansion', mrr_change, 0)) AS expansion_mrr,
    SUM(IF(event_type = 'contraction', mrr_change, 0)) AS contraction_mrr,
    SUM(IF(event_type = 'churn', mrr_change, 0)) AS churned_mrr,
    SUM(IF(event_type = 'reactivation', mrr_change, 0)) AS reactivated_mrr
  FROM mart_finance.mrr_events
  GROUP BY month
)
SELECT
  month, new_mrr, expansion_mrr, reactivated_mrr, contraction_mrr, churned_mrr,
  SUM(new_mrr + expansion_mrr + reactivated_mrr + contraction_mrr + churned_mrr)
    OVER (ORDER BY month) AS cumulative_mrr
FROM monthly
ORDER BY month