Detecting Incoming Satograms#

package satogram

import (
    "context"
    "encoding/hex"
    "errors"
    "fmt"
    "io"
    "strings"

    "github.com/lightningnetwork/lnd/lnrpc"
)

const (
    TLVMessage    uint64 = 34349334   // utf-8 message
    TLVRecipients uint64 = 6789998212 // single address OR comma-separated list
    TLVKeysend    uint64 = 5482373484 // keysend preimage
)

// CustodialFeeBps is your delivery fee in basis points. The actual rate is
// negotiated with the operator per wallet; this is just a placeholder.
const CustodialFeeBps = int64(1000)

type SettledSatogram struct {
    PaymentHash string
    AmtPaidSat  int64
    Message     string
    Recipients  []string // 1 entry for LNURL-pay, N for batch keysend
    IsKeysend   bool
}

func (s *Service) WatchInvoices(ctx context.Context) error {
    // Start from current tip so we don't replay history. For production,
    // persist the last add_index and resume from it on restart.
    info, err := s.lnd.ListInvoices(ctx, &lnrpc.ListInvoiceRequest{
        Reversed:       true,
        NumMaxInvoices: 1,
    })
    if err != nil {
        return fmt.Errorf("listing invoices: %w", err)
    }
    addIndex := info.LastIndexOffset

    sub, err := s.lnd.SubscribeInvoices(ctx, &lnrpc.InvoiceSubscription{
        AddIndex: addIndex,
    })
    if err != nil {
        return fmt.Errorf("subscribing: %w", err)
    }

    for {
        inv, err := sub.Recv()
        if errors.Is(err, io.EOF) {
            return nil
        }
        if err != nil {
            // In production: detect connection drops and resubscribe from
            // the last addIndex you saw.
            return fmt.Errorf("recv: %w", err)
        }
        if inv.State != lnrpc.Invoice_SETTLED {
            continue
        }
        s.handleSettled(ctx, inv)
    }
}

func (s *Service) handleSettled(ctx context.Context, inv *lnrpc.Invoice) {
    sg, ok := ExtractSatogram(inv)
    if !ok {
        // Not a Satogram. Could be a regular incoming payment.
        s.handleRegularInvoice(ctx, inv)
        return
    }

    // Split the paid amount across recipients minus your fee.
    credits := SplitAmount(sg.AmtPaidSat, sg.Recipients, CustodialFeeBps)

    for _, c := range credits {
        if err := s.users.CreditByLightningAddress(ctx, c.Address, c.Sats, CreditMeta{
            Source:      "satogram",
            PaymentHash: sg.PaymentHash,
            Message:     sg.Message,
        }); err != nil {
            s.log.Errorf("credit failed: addr=%s sats=%d err=%v", c.Address, c.Sats, err)
            continue
        }
        s.log.Infof("satogram: credited %s with %d sats (msg=%q)", c.Address, c.Sats, truncate(sg.Message, 60))
    }
}

func ExtractSatogram(inv *lnrpc.Invoice) (SettledSatogram, bool) {
    if len(inv.Htlcs) == 0 {
        return SettledSatogram{}, false
    }
    // Custom records live on the final-hop HTLC. For a keysend or any
    // single-HTLC payment, take the first. For MPP, scan them.
    var rawMsg, rawRecipients []byte
    for _, h := range inv.Htlcs {
        if h.CustomRecords == nil {
            continue
        }
        if v, ok := h.CustomRecords[TLVMessage]; ok && rawMsg == nil {
            rawMsg = v
        }
        if v, ok := h.CustomRecords[TLVRecipients]; ok && rawRecipients == nil {
            rawRecipients = v
        }
    }
    if rawRecipients == nil {
        return SettledSatogram{}, false
    }

    addresses := splitAndTrim(string(rawRecipients), ",")
    if len(addresses) == 0 {
        return SettledSatogram{}, false
    }
    return SettledSatogram{
        PaymentHash: hex.EncodeToString(inv.RHash),
        AmtPaidSat:  inv.AmtPaidSat,
        Message:     string(rawMsg), // safe to be empty
        Recipients:  addresses,
        IsKeysend:   inv.IsKeysend,
    }, true
}

type Credit struct {
    Address string
    Sats    int64
}

// SplitAmount takes a custodial cut (in bps) off the top and divides the
// remainder evenly across recipients. The remainder of integer division
// is kept by the custodian.
func SplitAmount(totalSats int64, recipients []string, feeBps int64) []Credit {
    if len(recipients) == 0 || totalSats <= 0 {
        return nil
    }
    cut := totalSats * feeBps / 10000
    net := totalSats - cut
    per := net / int64(len(recipients))
    out := make([]Credit, 0, len(recipients))
    for _, r := range recipients {
        out = append(out, Credit{Address: r, Sats: per})
    }
    return out
}

func splitAndTrim(s, sep string) []string {
    parts := strings.Split(s, sep)
    out := make([]string, 0, len(parts))
    for _, p := range parts {
        if p = strings.TrimSpace(p); p != "" {
            out = append(out, p)
        }
    }
    return out
}

func truncate(s string, n int) string {
    if len(s) <= n {
        return s
    }
    return s[:n] + "..."
}

#!/usr/bin/env python3
import json
import threading
from binascii import unhexlify
from pyln.client import Plugin
from pyln.proto.onion import TlvPayload

TLV_MESSAGE = 34349334
TLV_RECIPIENTS = 6789998212

plugin = Plugin()

# In-memory stash: payment_hash -> {"message": str, "recipients": [str]}.
# Replace with a persistent KV store (sqlite, plugin datastore) in production.
_pending = {}
_pending_lock = threading.Lock()


def _parse_satogram_tlvs(onion):
    """Return (message, [recipient_addresses]) or (None, None) if not a Satogram."""
    payload_hex = onion.get("payload")
    if not payload_hex:
        return None, None
    try:
        tlv = TlvPayload.from_bytes(unhexlify(payload_hex), skip_length=False)
    except Exception as e:
        plugin.log(f"failed to parse onion payload: {e}")
        return None, None

    message, recipients_raw = None, None
    for field in tlv.fields:
        # field.typenum may come back as int or str depending on pyln version.
        t = int(field.typenum) if str(field.typenum).isdigit() else field.typenum
        val = field.value if isinstance(field.value, (bytes, bytearray)) else str(field.value).encode()
        if t == TLV_MESSAGE:
            try:
                message = val.decode("utf-8")
            except UnicodeDecodeError:
                message = val.hex()
        elif t == TLV_RECIPIENTS:
            try:
                recipients_raw = val.decode("utf-8")
            except UnicodeDecodeError:
                recipients_raw = None

    if recipients_raw is None:
        return None, None
    recipients = [r.strip() for r in recipients_raw.split(",") if r.strip()]
    if not recipients:
        return None, None
    return message, recipients


@plugin.hook("htlc_accepted")
def on_htlc_accepted(onion, htlc, plugin, **kwargs):
    message, recipients = _parse_satogram_tlvs(onion)
    if recipients:
        payment_hash = htlc.get("payment_hash")
        with _pending_lock:
            _pending[payment_hash] = {"message": message or "", "recipients": recipients}
        plugin.log(f"satogram TLVs stashed for payment_hash={payment_hash} recipients={len(recipients)}")
    # Always let the HTLC continue; CLN handles settlement.
    return {"result": "continue"}


@plugin.hook("invoice_payment")
def on_invoice_payment(payment, plugin, **kwargs):
    payment_hash = payment.get("payment_hash") or payment.get("label")
    # `msat` arrives as a string like "5000msat" in some versions.
    msat_raw = payment.get("msat") or payment.get("amount_msat") or "0msat"
    msat = int(str(msat_raw).replace("msat", "").strip() or 0)
    amt_sat = msat // 1000

    with _pending_lock:
        sg = _pending.pop(payment_hash, None)

    if sg is None:
        # Not a Satogram, regular payment. Fall through to your normal credit path.
        return {"result": "continue"}

    credit_satogram(payment_hash, amt_sat, sg["message"], sg["recipients"])
    return {"result": "continue"}


# Replace with your custodial credit path.
CUSTODIAL_FEE_BPS = 1000  # placeholder; actual rate is negotiated with the operator per wallet

def credit_satogram(payment_hash, amt_sat, message, recipients):
    cut = amt_sat * CUSTODIAL_FEE_BPS // 10000
    net = amt_sat - cut
    per = net // len(recipients)
    for addr in recipients:
        # TODO: your real DB call, idempotent on payment_hash + addr.
        plugin.log(f"satogram credit: payment_hash={payment_hash} addr={addr} sats={per} msg={message[:60]!r}")


@plugin.init()
def init(options, configuration, plugin, **kwargs):
    plugin.log("satogram custodial plugin loaded")


plugin.run()

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use lightning::events::{Event, EventHandler, PaymentPurpose};
use lightning::ln::PaymentHash;
use lightning::ln::channelmanager::ChannelManager;

const TLV_MESSAGE: u64 = 34349334;
const TLV_RECIPIENTS: u64 = 6789998212;
const CUSTODIAL_FEE_BPS: u64 = 1000; // placeholder; actual rate is negotiated with the operator per wallet

#[derive(Clone)]
struct SatogramData {
    message: String,
    recipients: Vec<String>,
}

pub struct SatogramHandler<CM: AsRef<ChannelManager>> {
    channel_manager: CM,
    pending: Arc<Mutex<HashMap<[u8; 32], SatogramData>>>,
    // ... your DB handle, logger, etc.
}

impl<CM: AsRef<ChannelManager>> SatogramHandler<CM> {
    fn extract_satogram(custom_tlvs: &[(u64, Vec<u8>)]) -> Option<SatogramData> {
        let mut message = String::new();
        let mut recipients_raw: Option<String> = None;
        for (t, v) in custom_tlvs {
            match *t {
                TLV_MESSAGE => {
                    message = String::from_utf8_lossy(v).into_owned();
                }
                TLV_RECIPIENTS => {
                    recipients_raw = Some(String::from_utf8_lossy(v).into_owned());
                }
                _ => {}
            }
        }
        let csv = recipients_raw?;
        let recipients: Vec<String> = csv
            .split(',')
            .map(|s| s.trim().to_string())
            .filter(|s| !s.is_empty())
            .collect();
        if recipients.is_empty() {
            return None;
        }
        Some(SatogramData { message, recipients })
    }

    fn credit_satogram(&self, payment_hash: &PaymentHash, amount_msat: u64, sg: &SatogramData) {
        let amt_sat = amount_msat / 1000;
        let cut = amt_sat * CUSTODIAL_FEE_BPS / 10000;
        let net = amt_sat - cut;
        let per = net / sg.recipients.len() as u64;
        for addr in &sg.recipients {
            // TODO: your real credit call, keyed idempotently on (payment_hash, addr).
            log::info!(
                "satogram credit: payment_hash={} addr={} sats={} msg={:?}",
                hex::encode(payment_hash.0),
                addr,
                per,
                &sg.message.chars().take(60).collect::<String>(),
            );
        }
    }
}

impl<CM: AsRef<ChannelManager>> EventHandler for SatogramHandler<CM> {
    fn handle_event(&self, event: Event) {
        match event {
            Event::PaymentClaimable {
                payment_hash,
                amount_msat,
                purpose,
                onion_fields,
                ..
            } => {
                // Extract preimage from the purpose so we can claim.
                let preimage = match &purpose {
                    PaymentPurpose::Bolt11InvoicePayment {
                        payment_preimage: Some(p),
                        ..
                    } => *p,
                    PaymentPurpose::SpontaneousPayment(p) => *p,
                    _ => {
                        log::warn!("payment claimable without preimage; skipping");
                        return;
                    }
                };

                if let Some(of) = onion_fields.as_ref() {
                    if let Some(sg) = Self::extract_satogram(of.custom_tlvs()) {
                        self.pending.lock().unwrap().insert(payment_hash.0, sg);
                    }
                }

                self.channel_manager.as_ref().claim_funds(preimage);
            }
            Event::PaymentClaimed {
                payment_hash,
                amount_msat,
                ..
            } => {
                let sg = self.pending.lock().unwrap().remove(&payment_hash.0);
                if let Some(sg) = sg {
                    self.credit_satogram(&payment_hash, amount_msat, &sg);
                }
                // Else: not a Satogram, fall through to your normal credit path.
            }
            _ => {
                // Hand other events to your existing handler.
            }
        }
    }
}

// POST/GET /lnurlp/callback/:user?amount=<msat>&comment=<text>
app.get('/lnurlp/callback/:user', async (req, res) => {
  const username = req.params.user;
  const amountMsat = parseInt(req.query.amount, 10);
  const comment = req.query.comment || '';

  if (!Number.isFinite(amountMsat) || amountMsat < 1000) {
    return res.status(400).json({ status: 'ERROR', reason: 'bad amount' });
  }

  const user = await db.lookupUserByUsername(username);
  if (!user) {
    return res.status(404).json({ status: 'ERROR', reason: 'unknown user' });
  }

  // Eclair createinvoice. descriptionHash must be sha256 of the metadata
  // string served from /.well-known/lnurlp/<user>.
  const descriptionHash = metadataHashFor(username); // 32-byte hex
  const eclairResp = await fetch(`${ECLAIR_URL}/createinvoice`, {
    method: 'POST',
    headers: {
      Authorization: 'Basic ' + Buffer.from(':' + ECLAIR_PASSWORD).toString('base64'),
      'Content-Type': 'application/x-www-form-urlencoded',
    },
    body: new URLSearchParams({
      descriptionHash,
      amountMsat: String(amountMsat),
      expireIn: '600',
    }),
  });
  const invoice = await eclairResp.json();
  // invoice.serialized is the BOLT-11; invoice.paymentHash is hex.

  await db.recordPendingLnurlPayment({
    paymentHash: invoice.paymentHash,
    userId: user.id,
    amountMsat,
    comment,
  });

  res.json({ pr: invoice.serialized, routes: [] });
});

const WebSocket = require('ws');

const ws = new WebSocket(`${ECLAIR_WS_URL}/ws`, {
  headers: {
    Authorization: 'Basic ' + Buffer.from(':' + ECLAIR_PASSWORD).toString('base64'),
  },
});

ws.on('message', async (raw) => {
  let event;
  try { event = JSON.parse(raw); } catch (_) { return; }
  if (event.type !== 'payment-received') return;

  const paymentHash = event.paymentHash;
  const totalMsat = (event.parts || []).reduce(
    (sum, p) => sum + (p.amount || 0), 0
  );

  const pending = await db.takePendingLnurlPayment(paymentHash);
  if (!pending) return; // not a tracked LNURL-pay invoice, skip

  // For LNURL-pay Satograms, the "comment" we stored at callback time IS
  // the Satogram message (including the "📨 Satogram: " prefix the sender
  // service attaches before URL-encoding it).
  const message = pending.comment;
  const recipients = [pending.userId]; // Path A is always 1 recipient

  await creditSatogram(paymentHash, Math.floor(totalMsat / 1000), message, recipients);
});

async function creditSatogram(paymentHash, amtSat, message, recipients) {
  const feeBps = 1000n; // placeholder; actual rate is negotiated with the operator per wallet
  const total = BigInt(amtSat);
  const cut = (total * feeBps) / 10000n;
  const net = total - cut;
  const per = net / BigInt(recipients.length);
  for (const userId of recipients) {
    await db.creditUser({
      userId,
      sats: Number(per),
      paymentHash,
      source: 'satogram',
      memo: message,
    });
  }
}