Droplet Subscription Webhook Guide

curl -X POST "https://api.fluid.app/api/droplets" \
  -H "Authorization: Bearer C-YOUR_COMPANY_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "droplet": {
      "name": "Subscription Billing Synchronizer",
      "embed_url": "https://your-app.com/droplet/subscription-sync",
      "install_webhook_url": "https://your-app.com/webhooks/droplet/installed",
      "uninstall_webhook_url": "https://your-app.com/webhooks/droplet/uninstalled",
      "settings": {
        "marketplace_page": {
          "title": "Subscription Billing Synchronizer",
          "summary": "Synchronizes customer subscription billing dates to the 15th of each month",
          "logo_url": "https://your-app.com/logo.svg"
        },
        "details_page": {
          "title": "Subscription Billing Synchronizer",
          "summary": "Keeps all customer subscriptions on the same billing cycle",
          "logo_url": "https://your-app.com/big-logo.svg",
          "features": [
            {
              "name": "Automatic Synchronization",
              "summary": "Updates subscription billing dates",
              "details": "When a new subscription is created, other subscriptions for that customer are updated to bill on the 15th of the month"
            }
          ]
        }
      },
      "requested_scopes": ["main", "orders", "settings"]
    }
  }'

{
  "id": 12345,
  "identifier": "dropletinstallation_AbCdEf...",
  "name": "droplet_installed",
  "payload": {
    "event_name": "droplet_installed",
    "schema_version": 1,
    "schema_hash": "...",
    "contract_version": "v2",
    "company_id": 42,
    "resource_name": "DropletInstallation",
    "resource": "droplet",
    "event": "installed",
    "company": {
      "droplet_installation_uuid": "dri_AbCdEf...",
      "droplet_uuid": "drp_GhIjKl...",
      "fluid_company_id": 42,
      "fluid_shop": "acme",
      "name": "Acme Inc",
      "credentials": {
        "exchange_token": "dex_oneTimeOpaqueToken...",
        "exchange_token_expires_at": "2026-05-18T15:00:00Z",
        "exchange_endpoint": "/api/droplet_installations/exchange"
      }
    }
  },
  "timestamp": "2026-05-18T14:30:45Z"
}

class Webhooks::Droplet::InstalledController < ApplicationController
  skip_before_action :verify_authenticity_token

  def create
    raw_body = request.raw_post

    return head :request_timeout unless fresh_timestamp?
    return head :unauthorized unless valid_signature?(raw_body)

    # Acknowledge fast; do the exchange + persistence in a Sidekiq job so a
    # transient DB error does not consume the single-use exchange_token and
    # permanently break the install. See Step 4.
    Droplet::InstallJob.perform_later(JSON.parse(raw_body))
    head :ok
  end

  private

  def valid_signature?(raw_body)
    timestamp = request.headers["X-Fluid-Timestamp"]
    provided  = request.headers["X-Fluid-Signature"].to_s
    secret    = ENV.fetch("FLUID_DROPLET_WEBHOOK_SECRET") # the dws_ value from droplet creation

    expected = OpenSSL::HMAC.hexdigest("SHA256", secret, "#{timestamp}.#{raw_body}")
    ActiveSupport::SecurityUtils.secure_compare(expected, provided)
  end

  def fresh_timestamp?
    timestamp = request.headers["X-Fluid-Timestamp"].to_i
    (Time.now.to_i - timestamp).abs <= 300 # 5-minute tolerance
  end
end

{
  "droplet_installation": {
    "droplet_installation_uuid": "dri_...",
    "droplet_uuid": "drp_...",
    "fluid_company_id": 42,
    "fluid_shop": "acme"
  },
  "credentials": {
    "authentication_token": "dit_...",
    "webhook_verification_token": "wvt_...",
    "issued_at": "2026-05-18T14:30:45Z",
    "token_type": "bearer"
  }
}

class Droplet::InstallJob < ApplicationJob
  queue_as :webhooks

  def perform(install_payload)
    company = install_payload.fetch("payload").fetch("company")
    uuid    = company.fetch("droplet_installation_uuid")

    # Pre-create the tenant row with NULL tokens BEFORE the exchange. On a
    # Sidekiq retry after a partial failure, this find_or_create_by! resolves
    # to the existing row instead of duplicating, and the credentials check
    # below short-circuits if the previous attempt's exchange already wrote
    # them. Requires a unique index on droplet_tenants.droplet_installation_uuid.
    tenant = DropletTenant.find_or_create_by!(droplet_installation_uuid: uuid) do |t|
      t.company_id = company.fetch("fluid_company_id")
      t.fluid_shop = company.fetch("fluid_shop")
    end

    # Idempotent: if a prior attempt completed the exchange and wrote the
    # tokens, do nothing. The exchange_token is single-use, so we cannot
    # exchange again — but we also do not need to.
    return if tenant.installation_token.present?

    response = HTTParty.post(
      "https://api.fluid.app#{company.dig('credentials', 'exchange_endpoint')}",
      headers: { "Content-Type" => "application/json" },
      body:    { exchange_token: company.dig("credentials", "exchange_token") }.to_json,
      timeout: 20, # stay inside Fluid's 30-second lifecycle webhook window
    )
    raise "Exchange failed: #{response.code} #{response.body}" unless response.success?

    credentials = JSON.parse(response.body).fetch("credentials")
    tenant.update!(
      installation_token:         credentials.fetch("authentication_token"),       # dit_… — call Fluid back as this company
      webhook_verification_token: credentials.fetch("webhook_verification_token"), # wvt_… — HMAC key for business webhooks
    )
  end
end

# db/migrate/<timestamp>_create_droplet_tenants.rb
create_table :droplet_tenants do |t|
  t.bigint :company_id, null: false
  t.string :fluid_shop, null: false
  t.string :droplet_installation_uuid, null: false
  t.string :installation_token            # dit_… — populated by InstallJob after exchange
  t.string :webhook_verification_token    # wvt_… — populated by InstallJob after exchange
  t.timestamps
end
add_index :droplet_tenants, :company_id, unique: true                # one tenant per company
add_index :droplet_tenants, :fluid_shop, unique: true                # hot path for verifier lookup in Step 6
add_index :droplet_tenants, :droplet_installation_uuid, unique: true # idempotency key for InstallJob

# app/models/droplet_tenant.rb
class DropletTenant < ApplicationRecord
end

curl -X POST "https://api.fluid.app/api/company/webhooks" \
  -H "Authorization: Bearer dit_YOUR_INSTALLATION_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "webhook": {
      "resource": "subscription",
      "event": "started",
      "url": "https://your-app.com/webhooks/subscription-sync",
      "http_method": "post"
    }
  }'

{
  "id": 9876,
  "identifier": "commerce::subscription_aBcDeF...",
  "name": "subscription_started",
  "payload": {
    "event_name": "subscription_started",
    "schema_version": 3,
    "schema_hash": "...",
    "company_id": 42,
    "resource_name": "Commerce::Subscription",
    "resource": "subscription",
    "event": "started",
    "subscription": {
      "id": 12345,
      "subscription_token": "sb_AbCdEf...",
      "status": "active",
      "next_bill_date": "2026-06-01T00:00:00Z",
      "customer": { "id": 789, "email": "customer@example.com" },
      "subscription_plan": { "id": 10, "name": "Monthly Premium", "billing_interval_unit": "month" }
    }
  },
  "timestamp": "2026-05-18T14:30:45Z"
}

class Webhooks::SubscriptionSyncController < ApplicationController
  skip_before_action :verify_authenticity_token

  def create
    raw_body = request.raw_post

    return head :request_timeout unless fresh_timestamp?
    return head :unauthorized unless valid_signature?(raw_body, signing_key)

    payload = JSON.parse(raw_body)
    subscription = payload.dig("payload", "subscription")
    return head :ok unless subscription

    SubscriptionSyncJob.perform_later(payload)
    head :ok
  end

  private

  def signing_key
    # Look up the wvt_ value stored when we exchanged the install token in Step 4.
    # Use find_by (not find_by!) so unknown shops fail closed with 401 below,
    # rather than raising RecordNotFound and returning 404/500. Also guard
    # against a missing X-Fluid-Shop header so we never look up `IS NULL`.
    shop = request.headers["X-Fluid-Shop"]
    return nil if shop.blank?

    DropletTenant.find_by(fluid_shop: shop)&.webhook_verification_token
  end

  def valid_signature?(raw_body, secret)
    return false if secret.nil?

    timestamp = request.headers["X-Fluid-Timestamp"]
    provided  = request.headers["X-Fluid-Signature"].to_s

    expected = OpenSSL::HMAC.hexdigest("SHA256", secret, "#{timestamp}.#{raw_body}")
    ActiveSupport::SecurityUtils.secure_compare(expected, provided)
  end

  def fresh_timestamp?
    timestamp = request.headers["X-Fluid-Timestamp"].to_i
    (Time.now.to_i - timestamp).abs <= 300
  end
end

GET https://api.fluid.app/api/subscriptions?customer_id=789&status=active
Authorization: Bearer dit_YOUR_INSTALLATION_TOKEN

{
  "subscriptions": [
    {
      "id": 12345,
      "subscription_token": "sb_AbCdEf...",
      "status": "active",
      "next_bill_date": "2026-06-01T00:00:00Z",
      "subscription_plan": { "id": 10, ... },
      "customer": { "id": 789, ... },
      "currency": { "id": 1, ... },
      "variant": { "id": 333, "product": { ... } }
    }
  ],
  "meta": { "current_page": 1, "total_pages": 1, "total_count": 1, "stats": { ... } }
}

GET https://api.fluid.app/api/subscriptions/sb_AbCdEf...
Authorization: Bearer dit_YOUR_INSTALLATION_TOKEN

PATCH https://api.fluid.app/api/subscriptions/sb_AbCdEf...
Authorization: Bearer dit_YOUR_INSTALLATION_TOKEN
Content-Type: application/json

{
  "subscription": {
    "subscription_plan_id": 10,
    "customer_id": 789,
    "variant_id": 333,
    "address_id": 555,
    "payment_method_id": 444,
    "next_bill_date": "2026-06-15"
  }
}

class SubscriptionSynchronizer
  BILLING_DAY = 15

  def initialize(payload)
    @subscription = payload.fetch("subscription")
    @customer_id  = @subscription.dig("customer", "id")
    @company_id   = payload.fetch("company_id")
  end

  def synchronize
    other_subs = list_other_subscriptions
    target     = next_billing_date

    other_subs.each { |sub| update_bill_date(sub, target) }
  end

  private

  # Paginate through every active subscription for the customer — the list
  # endpoint defaults to 25 per page, and silently skipping later pages would
  # leave heavy customers' billing dates unsynchronized.
  def list_other_subscriptions
    page = 1
    all  = []

    loop do
      response = fluid.get("/api/subscriptions", query: { customer_id: @customer_id, status: "active", page: page })
      raise "List failed: #{response.code}" unless response.success?

      body = JSON.parse(response.body)
      all.concat(body.fetch("subscriptions"))

      break if page >= body.dig("meta", "total_pages").to_i

      page += 1
    end

    all.reject { |s| s["subscription_token"] == @subscription["subscription_token"] }
  end

  # The list endpoint's :api_index view omits address and payment_method,
  # so we re-fetch each subscription via show to get the full association set
  # required by the update endpoint.
  def fetch_full(token)
    response = fluid.get("/api/subscriptions/#{token}")
    raise "Show failed: #{response.code}" unless response.success?

    JSON.parse(response.body).fetch("subscription")
  end

  def update_bill_date(stub, target)
    sub = fetch_full(stub.fetch("subscription_token"))

    # UpdateAction requires payment_method_id and address_id as integers.
    # Some subscriptions (e.g. zero-price ones created without a stored
    # payment method) legitimately have nil here — patching them would 400.
    unless sub["payment_method"] && sub["address"]
      Rails.logger.info("Skipping #{sub['subscription_token']}: missing payment_method or address")
      return
    end

    body = {
      subscription: {
        subscription_plan_id: sub.dig("subscription_plan", "id"),
        customer_id:          sub.dig("customer", "id"),
        variant_id:           sub.dig("variant", "id"),
        address_id:           sub.dig("address", "id"),
        payment_method_id:    sub.dig("payment_method", "id"),
        next_bill_date:       target.strftime("%Y-%m-%d"),
      },
    }

    response = fluid.patch("/api/subscriptions/#{sub['subscription_token']}", body: body.to_json)

    if response.success?
      Rails.logger.info("Synced #{sub['subscription_token']} → #{target}")
    else
      Rails.logger.error("Sync failed for #{sub['subscription_token']}: #{response.code} #{response.body}")
    end
  end

  def next_billing_date
    today    = Date.current
    this_15  = Date.new(today.year, today.month, BILLING_DAY)
    today.day <= BILLING_DAY ? this_15 : this_15.next_month
  end

  def fluid
    @fluid ||= FluidClient.new(token: DropletTenant.find_by!(company_id: @company_id).installation_token)
  end
end

class FluidClient
  BASE_URL = "https://api.fluid.app"

  def initialize(token:)
    @headers = {
      "Authorization" => "Bearer #{token}",
      "Content-Type"  => "application/json",
    }
  end

  def get(path, query: {})
    HTTParty.get("#{BASE_URL}#{path}", headers: @headers, query: query, timeout: 30)
  end

  def patch(path, body:)
    HTTParty.patch("#{BASE_URL}#{path}", headers: @headers, body: body, timeout: 30)
  end
end

class SubscriptionSyncJob < ApplicationJob
  queue_as :webhooks

  def perform(webhook_body)
    SubscriptionSynchronizer.new(webhook_body.fetch("payload")).synchronize
  end
end

# config/routes.rb
Rails.application.routes.draw do
  post "/webhooks/droplet/installed",     to: "webhooks/droplet/installed#create"
  post "/webhooks/droplet/uninstalled",   to: "webhooks/droplet/uninstalled#create"
  post "/webhooks/subscription-sync",     to: "webhooks/subscription_sync#create"
end

require "rails_helper"

RSpec.describe Webhooks::SubscriptionSyncController, type: :request do
  let(:wvt)    { "wvt_test_value" }
  let(:tenant) { DropletTenant.create!(fluid_shop: "acme", company_id: 1, webhook_verification_token: wvt, installation_token: "dit_x") }
  let(:body)   { { id: 1, name: "subscription_started", payload: { subscription: { id: 99 } } }.to_json }
  let(:ts)     { Time.now.to_i.to_s }
  let(:sig)    { OpenSSL::HMAC.hexdigest("SHA256", wvt, "#{ts}.#{body}") }

  before { tenant }

  it "accepts a valid signature" do
    post "/webhooks/subscription-sync",
         params: body,
         headers: {
           "Content-Type"      => "application/json",
           "X-Fluid-Shop"      => "acme",
           "X-Fluid-Timestamp" => ts,
           "X-Fluid-Signature" => sig,
         }
    expect(response).to have_http_status(:ok)
  end

  it "rejects a stale timestamp before computing the signature" do
    post "/webhooks/subscription-sync",
         params: body,
         headers: {
           "Content-Type"      => "application/json",
           "X-Fluid-Shop"      => "acme",
           "X-Fluid-Timestamp" => (Time.now.to_i - 3600).to_s,
           "X-Fluid-Signature" => "deadbeef",
         }
    expect(response).to have_http_status(:request_timeout)
  end

  it "rejects an invalid signature" do
    post "/webhooks/subscription-sync",
         params: body,
         headers: {
           "Content-Type"      => "application/json",
           "X-Fluid-Shop"      => "acme",
           "X-Fluid-Timestamp" => ts,
           "X-Fluid-Signature" => "deadbeef",
         }
    expect(response).to have_http_status(:unauthorized)
  end
end

# Lifecycle webhook signing key — the dws_ secret generated when the droplet was created.
# Same for every install of your droplet; safe to keep in env.
FLUID_DROPLET_WEBHOOK_SECRET=dws_...

return if ProcessedWebhook.exists?(identifier: payload.fetch("identifier"))
ProcessedWebhook.create!(identifier: payload.fetch("identifier"))