1. KBKO VPN Bot — Telegram Auto-Sale Bot

What it is: A Telegram bot that sells VPN accounts automatically. Customer pays via WavePay, sends a payment screenshot, admin taps approve, and the bot creates the VPN account instantly through the Marzban API. Serves two Marzban panels (Spain + USA).

config.py — the sales rules

One file controls what the bot is allowed to sell. Changing the product line is a one-line edit, not a code hunt:

# /root/vpnbot/config.py
SOLD_PROTOCOLS = ["vless", "shadowsocks"]   # VMess/Trojan retired — one-line product change

PLANS = {
    "gb30":  {"gb": 30,  "days": 30, "price": 3000, "label": "30GB · 30 days"},
    "gb100": {"gb": 100, "days": 30, "price": 5000, "label": "100GB · 30 days"},
}

TRIAL_GB = 1
TRIAL_DAYS = 1
TRIAL_SERVER = "es"          # trials are Spain-only, USA is premium-only
REFERRAL_GB = 1              # +1GB per referred friend
ADMIN_IDS = [8500073574]

marzban.py — API client with token caching

Every API call needs a Bearer token. Logging in on every call would spam the panel's Telegram #Login alerts, so the client logs in once and reuses the token until the panel returns 401:

# /root/vpnbot/marzban.py
import requests, time

class Marzban:
    def __init__(self, base_url, username, password):
        self.base_url = base_url.rstrip("/")
        self.username = username
        self.password = password
        self._token = None

    def _login(self):
        r = requests.post(
            f"{self.base_url}/api/admin/token",
            data={"username": self.username, "password": self.password},
            verify=False, timeout=15)
        r.raise_for_status()
        self._token = r.json()["access_token"]

    def _req(self, method, path, retry=True, **kw):
        if not self._token:
            self._login()
        headers = kw.pop("headers", {})
        headers["Authorization"] = f"Bearer {self._token}"
        r = requests.request(method, f"{self.base_url}{path}",
                             headers=headers, verify=False,
                             timeout=20, **kw)
        if r.status_code == 401 and retry:
            self._login()                      # token expired -> one fresh login
            return self._req(method, path, retry=False, **kw)
        r.raise_for_status()
        return r.json() if r.text else None

    def create_user(self, username, gb, days):
        # Enforce the sales rule: ONLY VLESS Reality + Shadowsocks.
        # Even if the panel has VMess/Trojan inbounds, the user never gets them.
        proxies  = {p: {} for p in SOLD_PROTOCOLS}
        inbounds = {"vless": ["VLESS TCP REALITY"],
                    "shadowsocks": ["Shadowsocks TCP"]}
        return self._req("POST", "/api/user", json={
            "username": username,
            "proxies": proxies,
            "inbounds": inbounds,
            "data_limit": gb * 1024**3,
            "expire": int(time.time()) + days * 86400,
            # Defaults for every new user (both panels verified compatible):
            "flow": "xtls-rprx-vision",        # VLESS Reality flow
        })
        # NOTE: Shadowsocks method aes-256-gcm is set on the panel inbound itself.

bot.py — buy flow with server picker

Every command has both inline buttons and a working /cmd route. The buy flow starts with a server choice:

from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.ext import ContextTypes

async def buy(update: Update, context: ContextTypes.DEFAULT_TYPE):
    kb = [
        [InlineKeyboardButton("🇺🇸 USA · Premium ⭐", callback_data="srv_us")],
        [InlineKeyboardButton("🇪🇸 Spain",           callback_data="srv_es")],
    ]
    await update.message.reply_text(
        "Server ရွေးပါ / Choose server:",
        reply_markup=InlineKeyboardMarkup(kb))

async def on_server(update: Update, context: ContextTypes.DEFAULT_TYPE):
    q = update.callback_query
    await q.answer()
    server = q.data.replace("srv_", "")          # "us" or "es"
    context.user_data["server"] = server
    kb = [[InlineKeyboardButton(p["label"], callback_data=f"plan_{code}")]
          for code, p in PLANS.items()]
    await q.edit_message_text("Plan ရွေးပါ:",
                              reply_markup=InlineKeyboardMarkup(kb))

bot.py — WavePay screenshot → one-tap approve

Semi-auto payments: the customer sends a screenshot, the admin gets it with an Approve button, and approval creates the account immediately:

async def on_payment_photo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    uid = update.effective_user.id
    plan = context.user_data.get("plan")
    photo = update.message.photo[-1].file_id
    kb = [[InlineKeyboardButton("✅ Approve",
            callback_data=f"appr_{uid}_{plan}")]]
    await context.bot.send_photo(
        ADMIN_IDS[0], photo,
        caption=f"💰 Payment from {uid} · plan {plan}",
        reply_markup=InlineKeyboardMarkup(kb))
    await update.message.reply_text("⏳ Admin စစ်နေတယ်, ခဏစောင့်…")

async def on_approve(update: Update, context: ContextTypes.DEFAULT_TYPE):
    q = update.callback_query
    await q.answer()
    _, uid, plan = q.data.split("_", 2)
    server = user_server(uid)                    # stored at buy time
    mz = Marzban(**PANELS[server])               # Spain or USA panel
    p = PLANS[plan]
    try:
        u = mz.create_user(f"u{uid}", p["gb"], p["days"])
        save_order(uid, plan, server)            # SQLite: users/orders tables
        await context.bot.send_message(
            int(uid),
            f"✅ ရပြီ!\n\n{p['label']}\n\n"
            f"Subscription:\n<code>{u['subscription_url']}</code>\n\n"
            f"Happ / Hiddify ထဲထည့်, connect နှိပ် — ရပြီ 🚀",
            parse_mode="HTML")
        await q.edit_message_caption(caption="✅ Approved & created")
    except Exception as e:
        await q.edit_message_caption(caption=f"❌ Failed: {e}")

Deployment

# Systemd — Restart=always so a crash never needs a human
sudo tee /etc/systemd/system/vpnbot.service << 'EOF'
[Unit]
Description=KBKO VPN Bot
After=network.target
[Service]
Type=simple
User=root
WorkingDirectory=/root/vpnbot
ExecStart=/root/vpnbot/venv/bin/python bot.py
Restart=always
RestartSec=5
[Install]
WantedBy=multi-user.target
EOF
sudo systemctl daemon-reload && sudo systemctl enable --now vpnbot

Hard lessons: Marzban sends a Telegram #Login alert on every successful /api/admin/token — set the bot admin's telegram_id to NULL in the panel DB or your phone drowns in spam. Keep telegram_id on your human admin so real logins still alert. For monitoring, use non-authenticated reachability checks only.

2. Marzban VPN Infrastructure (Spain + USA)

What it is: Two production Marzban panels (Spain 31.70.155.75, USA 198.251.70.226) serving VLESS Reality and Shadowsocks via Xray-core. Hardened with iptables, fronted by nginx for public subscriptions.

Why VLESS is on port 443

Tested with Atom mobile data: Shadowsocks :1080 connected, but VLESS :2053, VMess :2080 and Trojan :2081 all timed out. The carrier filters uncommon ports. Port 443 (HTTPS) is never filtered, so VLESS Reality moved there:

# On the USA panel host — backup first, always
cp /var/lib/marzban/xray_config.json /var/lib/marzban/xray_config.json.bak2053

# In the Marzban panel: inbounds -> VLESS TCP REALITY -> port 2053 -> 443
# Then restart the container:
cd /opt/marzban && docker compose restart marzban

# Verify xray is actually listening on 443
ss -tlnp | grep ':443'        # expect: xray
# Existing customers must refresh their subscription to get the new port.

Firewall — panel locked to one IP

The Marzban dashboard (:8000) is only reachable from the Spain server. Everything else is dropped at the kernel level, before Marzban ever sees it:

# Run ON the USA Marzban host
iptables -A INPUT -i lo -j ACCEPT                                   # localhost always
iptables -A INPUT -p tcp --dport 8000 -s 31.70.155.75 -j ACCEPT      # Spain only
iptables -A INPUT -p tcp --dport 8000 -j DROP                       # everyone else
iptables-save > /etc/iptables/rules.v4                              # persist reboot

nginx — public subscriptions without VPN

Problem: customers buy VPN but needed VPN just to fetch their subscription link (both live on :8000). Fix: nginx on :8443 proxies only /sub/* to Marzban; /, /dashboard, /api all return 404. Note Marzban's :8000 is HTTPS with a self-signed cert, so the proxy disables upstream verification:

# /etc/nginx/sites-available/mm-sub-proxy
server {{
    listen 8443 ssl;
    server_name mm.kxant.eu.cc;
    ssl_certificate /etc/letsencrypt/live/mm.kxant.eu.cc/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/mm.kxant.eu.cc/privkey.pem;

    location /sub/ {{
        proxy_pass https://127.0.0.1:8000;   # Marzban is HTTPS (self-signed)
        proxy_ssl_verify off;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }}
    location / {{ return 404; }}              # dashboard stays Spain-only
}}

# Point new subscription URLs at the open port.
# /opt/marzban/.env:
XRAY_SUBSCRIPTION_URL_PREFIX="https://mm.kxant.eu.cc:8443"

# IMPORTANT: `docker compose restart` does NOT re-read .env.
# Recreate the container:
cd /opt/marzban && docker compose up -d
docker exec marzban-marzban-1 env | grep XRAY_SUBSCRIPTION_URL_PREFIX

Silence the bot-login Telegram spam

Marzban notifies the linked Telegram ID on every admin login — and the bot logs in constantly. Remove the link from the bot admin only; keep it on the human admin:

python3 -c "
import sqlite3
c = sqlite3.connect('/var/lib/marzban/db.sqlite3')
c.execute(\"UPDATE admins SET telegram_id = NULL WHERE username = 'vpnbot'\")
c.commit()
print(list(c.execute('SELECT username, telegram_id FROM admins')))
"
# -> [('kbko', 8500073574), ('vpnbot', None)]
# Bot logins: silent. Your manual logins: still alert. 

Hard lessons: Always cp xray_config.json{,.bak} before inbound edits. docker compose restart ignores .env changes — use up -d. Test ports from the user's real network (Atom), not from the server — server-side checks lie about carrier filtering.

3. PyHost Bot — Free Hosting Platform

What it is: A Telegram bot that gives users free Python hosting on the VPS. Users deploy scripts, the bot wraps each one in its own resource-capped systemd service. Premium tiers and one-time trial keys included.

Deployment model — one systemd service per user

Isolation is the whole design: a runaway script can only kill its own service, never the host or the VPN bot. Resource caps are non-negotiable:

import subprocess, textwrap

def deploy_script(user_id: int, filename: str):
    svc = f"pyhost-{{user_id}}"
    unit = textwrap.dedent(f"""
        [Unit]
        Description=PyHost user {{user_id}}
        After=network.target
        [Service]
        Type=simple
        User=nobody
        WorkingDirectory=/root/pyhostbot/apps/{{user_id}}
        ExecStart=/usr/bin/python3 {{filename}}
        Restart=on-failure
        # --- resource caps: one user can never take the server down ---
        MemoryMax=512M
        CPUQuota=100%
        [Install]
        WantedBy=multi-user.target
    """)
    open(f"/etc/systemd/system/{{svc}}.service", "w").write(unit)
    subprocess.run(["systemctl", "daemon-reload"], check=True)
    subprocess.run(["systemctl", "enable", "--now", svc], check=True)

Trial keys — PH-XXXXXXXX, one redeem per user

import secrets, sqlite3, time

def gen_key(days: int, max_uses: int, note: str = "") -> str:
    key = "PH-" + secrets.token_hex(4).upper()          # e.g. PH-9F3A2C1D
    db.execute("INSERT INTO trial_keys(key,days,max_uses,used,note) VALUES(?,?,?,?,?)",
               (key, days, max_uses, 0, note))
    db.commit()
    return key

def redeem_key(user_id: int, key: str) -> str:
    row = db.execute("SELECT days,max_uses,used FROM trial_keys WHERE key=?",
                     (key,)).fetchone()
    if not row:                       return "❌ Key မမှန်ဘူး"
    if row["used"] >= row["max_uses"]: return "❌ ဒီ key ကုန်သွားပြီ"
    if db.execute("SELECT 1 FROM redemptions WHERE key=? AND user_id=?",
                  (key, user_id)).fetchone():
        return "❌ နင် ဒီ key သုံးပြီးပြီ (တစ်ယောက် တစ်ခါပဲ)"
    db.execute("UPDATE trial_keys SET used=used+1 WHERE key=?", (key,))
    db.execute("INSERT INTO redemptions(key,user_id) VALUES(?,?)", (key, user_id))
    # grant biggest-plan premium for `days`
    db.execute("INSERT OR REPLACE INTO premiums(user_id,plan,expire) VALUES(?,?,?)",
               (user_id, "pro", int(time.time()) + row["days"] * 86400))
    db.commit()
    return f"✅ {{row['days']}} ရက် Pro ရပြီ!"

/admin panel — tappable commands

Admins get buttons, not a memorized command list. Every button maps to the same /cmd handler so both paths always work:

async def admin_panel(update, context):
    kb = [
        [InlineKeyboardButton("📋 Plans",       callback_data="adm_plans")],
        [InlineKeyboardButton("💎 Premiums",    callback_data="adm_premiums")],
        [InlineKeyboardButton("🔑 Keys",        callback_data="adm_keys")],
        [InlineKeyboardButton("➕ /addpremium", callback_data="adm_addpremium")],
    ]
    await update.message.reply_text("Admin panel:",
        reply_markup=InlineKeyboardMarkup(kb))
# Rule: every callback_data "adm_X" calls the same function as /X.

Hard lessons: Premium ≠ admin — free deploy/renew stayed admin-only even for premium users. Cap memory per deployment or one infinite loop OOMs the whole VPS. Fake test doubles must mirror the real library API exactly, or tests pass and production crashes.

4. VidDrop — Telegram Video Downloader

What it is: Paste any video link → get quality buttons + MP3 option → the file arrives in chat. Powered by yt-dlp, capped at 2 concurrent downloads so the VPS never chokes.

Format listing → inline buttons

import yt_dlp, asyncio

def list_formats(url: str):
    with yt_dlp.YoutubeDL({{"quiet": True}}) as ydl:
        info = ydl.extract_info(url, download=False)
    # pick one stream per height, skip absurd sizes
    seen, out = set(), []
    for f in info.get("formats", []):
        h = f.get("height")
        if h and h not in seen and h <= 1080:
            seen.add(h)
            out.append((h, f["format_id"]))
    return info.get("title", "video"), sorted(out)

async def on_link(update, context):
    title, fmts = await asyncio.to_thread(list_formats, update.message.text)
    kb = [[InlineKeyboardButton(f"🎬 {{h}}p", callback_data=f"dl_{{fid}}_{{h}}")]
          for h, fid in fmts]
    kb.append([InlineKeyboardButton("🎵 MP3", callback_data="dl_mp3")])
    context.user_data["dl_url"] = update.message.text
    await update.message.reply_text(f"{{title}}\nQuality ရွေးပါ:",
        reply_markup=InlineKeyboardMarkup(kb))

Download with a semaphore (max 2 at once)

SEMAPHORE = asyncio.Semaphore(2)

async def on_quality(update, context):
    q = update.callback_query
    await q.answer()
    url = context.user_data["dl_url"]
    async with SEMAPHORE:                       # never more than 2 downloads
        if q.data == "dl_mp3":
            opts = {{"format": "bestaudio/best",
                     "postprocessors": [{{"key": "FFmpegExtractAudio",
                                          "preferredcodec": "mp3"}}]}}
        else:
            _, fid, h = q.data.split("_")
            opts = {{"format": fid, "max_filesize": 50 * 1024 * 1024}}
        opts.update({{"outtmpl": "/tmp/viddrop/%(id)s.%(ext)s", "quiet": True}})
        fname = await asyncio.to_thread(download, url, opts)
        await q.message.reply_document(open(fname, "rb"))
        os.remove(fname)                        # don't fill the disk

def download(url, opts):
    with yt_dlp.YoutubeDL(opts) as ydl:
        info = ydl.extract_info(url, download=True)
        return ydl.prepare_filename(info)

Hard lessons: TikTok blocks datacenter IPs — the fix is the user's own logged-in cookies.txt passed via --cookies; never print cookie contents. 50MB is Telegram's bot file cap — enforce max_filesize before downloading, not after.

5. DubLab — AI Video Dubbing Website

What it is: Upload a short video (≤5 min / 250MB) → pick EN↔MM + male/female voice → get a dubbed MP4. Pipeline: faster-whisper transcribes → Gemini translates → Edge-TTS speaks (Burmese voices work!) → FFmpeg muxes. Plus an SRT translator tab.

Step 1 — transcribe with faster-whisper

from faster_whisper import WhisperModel
model = WhisperModel("small", device="cpu", compute_type="int8")

def transcribe(path: str):
    segments, _ = model.transcribe(path, language="en")
    return [{{"start": s.start, "end": s.end, "text": s.text.strip()}}
            for s in segments if s.text.strip()]
# Dependency pins that actually work together (faster-whisper 1.2.1):
#   av==15.1.0  +  huggingface_hub<1
# av 19.x removed the `metadata_errors` kwarg faster-whisper passes -> TypeError.
# av 14.x has no cp312 wheel -> build fails. 15.1.0 is the sweet spot.

Step 2 — translate with Gemini key rotation

Free Gemini quota is tiny (~10–15 req/day/key ≈ 1–2 videos/day). Rotate through keys; when one hits quota, try the next:

import google.generativeai as genai, itertools

KEYS = open("/root/dublab/.env").read().split("GEMINI_API_KEYS=")[1].split(",")
key_cycle = itertools.cycle(k.strip() for k in KEYS)

def translate(text: str, target: str) -> str:
    last_err = None
    for _ in range(len(KEYS)):                 # try each key once
        genai.configure(api_key=next(key_cycle))
        try:
            m = genai.GenerativeModel("gemini-2.0-flash")
            r = m.generate_content(
                f"Translate to {{target}}, return ONLY the translation: {{text}}")
            return r.text.strip()
        except Exception as e:
            last_err = e                        # quota error -> next key
    raise last_err

Step 3 — dub with Edge-TTS (Burmese voices!)

import edge_tts, asyncio

VOICES = {{
    "mm-male":   "my-MM-ThihaNeural",
    "mm-female": "my-MM-NilarNeural",
    "en-male":   "en-US-ChristopherNeural",
    "en-female": "en-US-AriaNeural",
}}

async def speak(text: str, voice: str, out: str):
    await edge_tts.Communicate(text, voice).save(out)

# Per segment, then stretch/compress each clip to the segment's
# timestamp window with FFmpeg `atempo` so lips roughly match.

Step 4 — mux with FFmpeg

# dubbed.mp3 = concatenated per-segment clips, original kept quiet underneath
subprocess.run([
    "ffmpeg", "-y",
    "-i", "original.mp4",
    "-i", "dubbed.mp3",
    "-filter_complex",
        "[0:a]volume=0.25[orig];[1:a]adelay=0[db];[orig][db]amix=inputs=2[a]",
    "-map", "0:v", "-map", "[a]",
    "-c:v", "copy", "-c:a", "aac",
    "output.mp4"], check=True)

Serving — waitress behind nginx, iOS-safe uploads

# waitress on 127.0.0.1 ONLY (never expose the app server directly)
# /etc/systemd/system/dublab.service:
# ExecStart=/root/dublab/venv/bin/waitress-serve --listen=127.0.0.1:5000 app:app

/* iOS Safari bug: fetch() POST with FormData throws
   "TypeError: The string did not match the expected pattern" (WebKit bug).
   Fix: XMLHttpRequest for uploads, fetch only for JSON polling. */
function uploadFile(file) {
  return new Promise((resolve, reject) => {
    const xhr = new XMLHttpRequest();
    const fd = new FormData();
    fd.append("video", file);
    xhr.open("POST", "/api/dub");
    xhr.onload = () => resolve(JSON.parse(xhr.responseText));
    xhr.onerror = reject;
    xhr.send(fd);
  });
}

Hard lessons: Jobs in process memory die on restart — an old job ID becomes invalid after any service restart. Rate-limit uploads (10/hour per IP) or one user eats the whole Gemini quota. The monitor must never auto-reopen the site against an explicit close order.

6. MyCinema — Movie Streaming Website

What it is: "Watch edition" streaming site on a Windows VPS — search any title, pick a server, watch instantly. 6,487 movies + 1,330 TV shows, all stream-resolvable. No download links.

Player — server URL templates

Three main servers plus 229 MoviesHUB per-file hub servers as extras. The app only swaps the iframe src:

const SERVERS = [
  {{ name: "Server 1 · VidSrc",  movie: id => `https://vidsrc.to/embed/movie/${{id.imdb}}`,
                                 tv: (id,s,e) => `https://vidsrc.to/embed/tv/${{id.imdb}}/${{s}}/${{e}}` }},
  {{ name: "Server 2 · MoviesAPI", movie: id => `https://moviesapi.to/movie/${{id.tmdb}}`,
                                   tv: (id,s,e) => `https://moviesapi.to/tv/${{id.tmdb}}-${{s}}-${{e}}` }},
  {{ name: "Server 3 · VidLink",  movie: id => `https://vidlink.pro/movie/${{id.tmdb}}`,
                                  tv: (id,s,e) => `https://vidlink.pro/tv/${{id.tmdb}}/${{s}}/${{e}}` }},
];

function play(ids, serverIdx = 0, season = 1, ep = 1) {{
  const s = SERVERS[serverIdx];
  const url = ids.imdb && serverIdx === 0
      ? s.movie(ids) : s.tv(ids, season, ep);
  document.getElementById("player").src = url;
  saveResume(ids, serverIdx, season, ep);   // 30-day TTL in localStorage
}}

Search via TMDB

const TMDB = "https://api.themoviedb.org/3";
async function search(q) {{
  const r = await fetch(`${{TMDB}}/search/multi?api_key=${{API_KEY}}&query=${{encodeURIComponent(q)}}`);
  const j = await r.json();
  return j.results.filter(x => x.media_type === "movie" || x.media_type === "tv");
}}
// Catalog is pre-built (movies.json / tvshows.json) from TMDB —
// only titles resolvable by at least one server are included.

SRT subtitles — upload, convert, serve with CORS

VidSrc honors external subtitles if you hand it a VTT URL. The server converts SRT→VTT and serves it CORS-enabled:

# server.py
def srt_to_vtt(srt_text: str) -> str:
    vtt = "WEBVTT\n\n" + srt_text.replace(",", ".")
    return vtt   # timestamps: 00:00:01,000 -> 00:00:01.000

@app.route("/sub/<name>.vtt")
def serve_vtt(name):
    resp = make_response(open(f"subs/{{name}}.vtt").read())
    resp.headers["Content-Type"] = "text/vtt"
    resp.headers["Access-Control-Allow-Origin"] = "*"   # iframe needs this
    return resp
// player: https://vidsrc.to/embed/movie/{imdb}?sub=https://cinema.kxant.org/sub/x.vtt

Resume-play (item + server, not timestamp)

function saveResume(ids, serverIdx, s, e) {
  localStorage.setItem("mycinema_resume", JSON.stringify({
    ids, serverIdx, s, e, ts: Date.now()
  }));
}
function getResume() {
  const r = JSON.parse(localStorage.getItem("mycinema_resume") || "null");
  if (!r || Date.now() - r.ts > 30*86400*1000) return null;  // 30-day TTL
  return r;
}
// Exact playback position is impossible: cross-origin iframes
// can't report currentTime. Item + server + episode is the best we can do.

Windows deployment gotcha

# Scheduled Tasks run as SYSTEM with a DIFFERENT PATH than your admin shell.
# A bare `python` in the task action dies instantly with no log.
# ALWAYS resolve the full exe path:
$pyExe  = (Get-Command python3.14).Source
$action = New-ScheduledTaskAction -Execute $pyExe -Argument "C:\mycinema\server.py"
Register-ScheduledTask -TaskName "MyCinema" -Action $action `
  -Trigger (New-ScheduledTaskTrigger -AtStartup) -User "SYSTEM" -RunLevel Highest

Hard lessons: VidSrc hard-blocks sandboxed iframes — the ad-block sandbox toggle only applies to moviesapi.to. Pre-build the catalog; live TMDB search alone returns titles no server can play.

7. n8n Notification Hub

What it is: Self-hosted n8n on Spain that watches everything — website uptime (5 min), Marzban panel reachability (15 min), systemd services (15 min), SSL expiry (daily) — and pings Telegram on problems. All alerts go through the VPN bot to the admin.

Run it capped and isolated

docker run -d --name n8n --restart unless-stopped \
  --memory=1g --cpus=1 \
  -p 127.0.0.1:5678:5678 \
  -v n8n_data:/home/node/.n8n \
  n8nio/n8n
# nginx: n8n.kbko.qzz.io -> 127.0.0.1:5678 (Let's Encrypt)

Docker-subnet DNS fix

Marzban's dnstt-server hijacks UDP/53, which also breaks DNS for Docker containers. Add a scoped exception for the Docker subnet only — VPN-client DNS stays untouched:

# Backup first!
cp /etc/iptables/rules.v4 /etc/iptables/rules.v4.bak_n8n

# Allow the Docker subnet to reach real DNS, before the hijack rule:
iptables -t nat -I PREROUTING 1 -s 172.18.0.0/16 -p udp --dport 53 -j ACCEPT
iptables-save > /etc/iptables/rules.v4

Workflow pattern — every check looks like this

{
  "nodes": [
    { "name": "Every 5 min",
      "type": "n8n-nodes-base.scheduleTrigger",
      "parameters": { "rule": { "interval": [{ "field": "minutes", "minutesInterval": 5 }] } } },
    { "name": "Check site",
      "type": "n8n-nodes-base.httpRequest",
      "parameters": { "url": "https://dublab.kbko.qzz.io/health",
                      "options": { "timeout": 10000 } } },
    { "name": "Is it down?",
      "type": "n8n-nodes-base.if",
      "parameters": { "conditions": {
        "number": [{ "value1": "={{$json.statusCode}}", "operation": "notEqual", "value2": 200 }] } } },
    { "name": "Alert admin",
      "type": "n8n-nodes-base.telegram",
      "parameters": { "chatId": "8500073574",
        "text": "🚨 dublab.kbko.qzz.io down! HTTP {{$json.statusCode}}" } }
  ],
  "connections": {
    "Every 5 min": { "main": [[{ "node": "Check site" }]] },
    "Check site":  { "main": [[{ "node": "Is it down?" }]] },
    "Is it down?": { "main": [[{ "node": "Alert admin" }], []] }
  }
}
// Same skeleton for: Marzban panel :8000 (15 min), systemd via SSH node,
// SSL expiry via `openssl s_client` in an Execute Command node (daily).

Hard lessons: For Marzban monitoring use non-authenticated reachability checks — a credentialed /api/admin/token login fires a user-visible Telegram #Login alert. Cap n8n at 1GB RAM; it will happily eat everything with enough workflows.

8. Outline Manager — Phone VPS Dashboard

What it is: A mobile-friendly Flask dashboard to run the whole VPN infrastructure from a phone — Outline keys (QR + link), Marzban users, 3xUI inbounds, SSH tunnel + SlowDNS setup, Cloudflare DNS editor, one-click installers, backup/restore. Secrets are Fernet-encrypted at rest, admin-only.

Secrets — Fernet, never plaintext

from cryptography.fernet import Fernet
import os

KEY_PATH = "/root/outline-manager/.fernet.key"   # chmod 600
if not os.path.exists(KEY_PATH):
    open(KEY_PATH, "wb").write(Fernet.generate_key())
    os.chmod(KEY_PATH, 0o600)
fernet = Fernet(open(KEY_PATH, "rb").read())

def enc(s: str) -> str:  return fernet.encrypt(s.encode()).decode()
def dec(t: str) -> str:  return fernet.decrypt(t.encode()).decode()

# Stored in SQLite as enc(api_key); decrypted only in memory at request time.
# Even a DB leak reveals nothing without the key file.

Outline key → QR + link

import requests, qrcode, io, base64

def outline_keys():
    api = dec(get_setting("outline_api_url"))   # https://host:port/SECRET/
    r = requests.get(api + "access-keys/", verify=False, timeout=15)
    return r.json()["accessKeys"]

@app.route("/outline/new", methods=["POST"])
def outline_new():
    api = dec(get_setting("outline_api_url"))
    r = requests.post(api + "access-keys/", verify=False, timeout=15)
    key = r.json()
    # QR for the phone camera:
    img = qrcode.make(key["accessUrl"])
    buf = io.BytesIO(); img.save(buf, format="PNG")
    qr_b64 = base64.b64encode(buf.getvalue()).decode()
    return {"link": key["accessUrl"], "qr": qr_b64, "id": key["id"]}

Cloudflare DNS without the dashboard

The Cloudflare dashboard is bot-walled in automation environments. Skip it — the API works fine with a scoped token:

# Ask the user to create an "Edit zone DNS" token scoped to the zone,
# then call the API directly:
CF = "https://api.cloudflare.com/client/v4"
H = {"Authorization": f"Bearer {dec(get_setting('cf_token'))}"}

def dns_set(name: str, ip: str):
    zone = requests.get(f"{CF}/zones?name=kxant.org", headers=H).json()["result"][0]["id"]
    recs = requests.get(f"{CF}/zones/{zone}/dns_records?name={name}",
                        headers=H).json()["result"]
    data = {"type": "A", "name": name, "content": ip, "ttl": 300, "proxied": False}
    if recs:
        requests.put(f"{CF}/zones/{zone}/dns_records/{recs[0]['id']}",
                     headers=H, json=data)
    else:
        requests.post(f"{CF}/zones/{zone}/dns_records", headers=H, json=data)

Hard lessons: SlowDNS needs a non-interactive installer — the form takes tunnel domain + SSH port and shows the public key, no SSH session required. Admin password changes must invalidate existing sessions.

9. Movie Scraper Bot (Telethon)

What it is: An in-bot movie scraper — searches Telegram channels for movies and serves results inside the bot. Built on Telethon (user client, not bot API) so it can read channel history.

Client setup — secrets never in code

# /root/moviebot/.tg_env  (chmod 600) — the ONLY place secrets live:
# API_ID=123456
# API_HASH=abcdef123456
# (first run does interactive login, then moviebot.session persists)

from telethon import TelegramClient
import os

def load_env(path):
    d = {{}}
    for line in open(path):
        if "=" in line and not line.startswith("#"):
            k, v = line.strip().split("=", 1)
            d[k] = v
    return d

env = load_env("/root/moviebot/.tg_env")
client = TelegramClient("/root/moviebot/moviebot", env["API_ID"], env["API_HASH"])

async def search_movies(query: str, limit=10):
    results = []
    async for msg in client.iter_messages("@movie_channel", search=query, limit=limit):
        if msg.video or msg.document:
            results.append({{"id": msg.id, "text": msg.text or "", "date": msg.date}})
    return results

Dead-token crash loop — the fix

# If the BotFather token dies, the service crash-loops every few seconds.
# That once helped OOM-freeze the whole Spain server (with musicbot).
# Guard: stop + disable until a fresh token exists.
#   systemctl stop moviebot && systemctl disable moviebot
# Then paste a NEW token from @BotFather into .tg_env and re-enable.

Hard lessons: A dead token doesn't fail quietly — Telethon retries forever, eating CPU/RAM in a tight loop. Combined with zero swap, two such bots froze the entire Spain VPS. Monitor for crash-looping services, not just stopped ones.

10. OSINT Bot

What it is: A Telegram bot for quick lookups: /ip /email /user /auto /inn /all /proxycheck. Built for speed — one message in, one answer out.

Proxy handling — file format + graceful fallback

Proxies die constantly. The bot reads proxy.txt (host:port:user:pass per line) and falls back to a direct connection instead of hard-failing:

import random, requests

def load_proxies(path="/root/info/proxy.txt"):
    proxies = []
    for line in open(path):
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        parts = line.split(":")          # host:port:user:pass
        if len(parts) == 4:
            h, p, u, pw = parts
            proxies.append(f"http://{{u}}:{{pw}}@{{h}}:{{p}}")
    return proxies

PROXIES = load_proxies()

def fetch(url, timeout=15):
    last_err = None
    for px in random.sample(PROXIES, min(3, len(PROXIES))):
        try:
            return requests.get(url, proxies={{"http": px, "https": px}},
                                timeout=timeout).json()
        except Exception as e:
            last_err = e                  # proxy dead -> try next
    # All proxies failed: direct fallback (better than nothing)
    return requests.get(url, timeout=timeout).json()

Command skeleton

async def cmd_ip(update, context):
    target = " ".join(context.args)
    if not target:
        return await update.message.reply_text("Usage: /ip 1.2.3.4")
    try:
        d = await asyncio.to_thread(fetch, f"http://ip-api.com/json/{{target}}")
        await update.message.reply_text(
            f"🌐 {{d.get('query')}}\n"
            f"📍 {{d.get('city')}}, {{d.get('country')}}\n"
            f"🏢 {{d.get('isp')}}\n"
            f"🕒 {{d.get('timezone')}}")
    except Exception as e:
        await update.message.reply_text(f"❌ Lookup failed: {{e}}")
# Same shape for /email /user /inn — one fetcher, one formatter each.

Hard lessons: The original crash was a proxy-format mismatch (host:port:user:pass vs code expecting something else) plus missing imports — always validate the proxy file on startup and log which line failed.

11. VPN Reseller Bot

What it is: A separate wholesale panel for VPN resellers — 30% off retail, reseller manages their own customers. Terms: ≥10 accounts/month, never below retail, WavePay in advance or weekly.

Isolation — the non-negotiable rule

# Every bot gets its OWN everything. Sharing is how outages happen.
#   Token:    @kbkovpnreseller_bot  (different BotFather token)
#   Dir:      /root/resellerbot      (not /root/vpnbot)
#   Service:  resellerbot.service    (not vpnbot.service)
#   DB:       /root/resellerbot/reseller.db (separate SQLite)
#   Admin:    same human, but admin/user command namespaces never overlap

Wholesale provisioning

RESELLER_DISCOUNT = 0.30   # 30% off retail

def wholesale_price(retail_ks: int) -> int:
    return int(retail_ks * (1 - RESELLER_DISCOUNT))

async def reseller_buy(update, context):
    # Reseller buys at wholesale; bot creates the Marzban user
    # exactly like the main bot, but tagged with reseller_id
    # so reports and renewals route to the right person.
    uid = update.effective_user.id
    if not is_reseller(uid):
        return await update.message.reply_text("❌ Reseller only.")
    ...

15-minute health watcher (cron)

# crontab -e:
*/15 * * * * /root/watchers/resellerbot-watch.sh >> /var/log/resellerbot-watch.log 2>&1

#!/bin/bash
# /root/watchers/resellerbot-watch.sh
if ! systemctl is-active --quiet resellerbot; then
  MSG="🚨 Reseller bot ပြဿနာရှိတယ်\n\n• service ရပ်နေတယ်\nစစ်ပေးပါဦး။"
elif [ "$(journalctl -u resellerbot --since '20 minutes ago' -o cat | grep -ci traceback)" -gt 0 ]; then
  MSG="🚨 Reseller bot ပြဿနာရှိတယ်\n\n• Traceback error တွေ့တယ်\nစစ်ပေးပါဦး။"
elif [ "$(journalctl -u resellerbot --since '20 minutes ago' -o cat | grep -c getUpdates)" -eq 0 ]; then
  MSG="🚨 Reseller bot ပြဿနာရှိတယ်\n\n• Telegram နဲ့ အဆက်ပြတ်နေတယ်\nစစ်ပေးပါဦး။"
fi
[ -n "$MSG" ] && send_telegram "$MSG"   # silent when healthy

Hard lessons: The onboarding kit (terms card + 30% pricing table + provisioning checklist + tested Spain→USA migration steps) is reusable for every new reseller — build it once. Never mix admin, reseller and user controls.

12. Health Watchers & Automation

What it is: The monitoring layer keeping everything alive — hourly VPN health checks, 15-minute service watchers, a freeze auto-recovery, quota watchers, and a daily preventive reboot of all three servers.

The watcher pattern — 2 consecutive failures before alerting

This one rule kills 99% of false alarms. A single failed check means nothing; two in a row means something:

#!/bin/bash
# /root/watchers/vpn-server-health-watch.sh (simplified)
STATE=/root/watchers/state/health.json
URL="https://panel.kbko.qzz.io/"

was_down=$(jq -r .was_down "$STATE")
fails=$(jq -r .fails "$STATE")

if curl -sk --max-time 10 "$URL" | grep -q "Marzban"; then
    fails=0
    if [ "$was_down" = true ]; then
        send_telegram "✅ Panel ပြန်ကောင်းသွားပါပြီ"
    fi
    was_down=false
else
    fails=$((fails + 1))
    if [ "$fails" -ge 2 ] && [ "$was_down" = false ]; then
        # Second consecutive failure -> real alert, with fix buttons
        send_telegram "🚨 Panel down (2 checks)\n\n✅ ပြင်ပါ / ⏭️ ထားလိုက်မယ်"
        was_down=true
    fi
fi
echo "{{\"was_down\": $was_down, \"fails\": $fails}}" > "$STATE"
# Silent when healthy. Re-alert only after 2 MORE consecutive failures.
# Fix-approval: next run fixes ONLY if the ✅ button was tapped.

Spain freeze auto-recovery

Standing order: "freezeဖြစ်တိုင်း မင်းဘာမင်းစစ်ပြီး လိုအပ်တာ တခါထဲ လုပ်လိုက်ပါ". Every 15 min: if both SSH and the web check fail twice in a row, reboot Spain via the IONOS panel (max once per incident), then verify and report in Burmese:

# Pseudo-logic of spain-freeze-auto-recovery (15-min cron):
ssh_ok=$(timeout 10 ssh spain "echo ok" 2>/dev/null)
web_ok=$(curl -sk --max-time 10 https://n8n.kbko.qzz.io/ -o /dev/null -w "%{{http_code}}")

if [ -z "$ssh_ok" ] && [ "$web_ok" != "200" ]; then
    fails=$((fails + 1))
else
    fails=0; rebooted_this_incident=false
fi

if [ "$fails" -ge 2 ] && [ "$rebooted_this_incident" = false ]; then
    ionos_panel_restart "31.70.155.75"   # Restart action, that server ONLY
    rebooted_this_incident=true
    sleep 180; verify_services           # dublab, vpnbot, nginx, marzban
    send_telegram "🔄 Spain freeze → reboot လုပ်ပြီး, service တွေပြန်ကောင်း ✅"
fi

Root cause found (Oct 2025) — and fixed

# NOT a payment issue (contract clean). The real cause:
#  - moviebot (dead BotFather token) + musicbot (login pending)
#    crash-looped every few seconds
#  - 3.8 GB RAM + ZERO swap -> memory-pressure stalls
#    (panel showed RUNNING, but SSH + web were dead)
#
# Fix:
systemctl stop moviebot musicbot
systemctl disable moviebot musicbot
fallocate -l 2G /swapfile && chmod 600 /swapfile
mkswap /swapfile && swapon /swapfile
echo "/swapfile none swap sw 0 0" >> /etc/fstab   # persist reboot

Daily preventive restart (07:00)

# 0 7 * * * /root/watchers/daily-restart.sh
# Spain:  reboot via SSH, verify dublab/resellerbot/vpnbot/nginx/marzban
# USA:    reboot via SSH, verify marzban container
# Windows: restart via IONOS panel API, verify IP + status On
# Report: one Burmese summary, silent if all green.

Hard lessons: Alerts go only to the admin Telegram ID — never to customers. No proactive watchers unless explicitly requested; the approved set is enough.

🛠️ Universal Patterns

Rules applied to every project, learned the hard way:

1. Systemd everywhere: Restart=always, RestartSec=5, enable --now. A service that can't survive a reboot isn't done.
2. Secrets in .env (chmod 600) — never in code, chat, logs, or screenshots. Bot tokens only in env.
3. Backup before changes: cp file{,.bak$(date +%Y%m%d)}. The 30 seconds it takes has saved hours.
4. Create-and-verify before delete — especially user accounts and server migrations. New first, delete old only after the new one works.
5. Inline buttons AND /cmd routes — every bot command, both, always. No exceptions.
6. Burmese-first messages — Unicode only (Zawgyi looks broken), short, Telegram-friendly.
7. Free-only — every project runs on free tiers. Never suggest paid plans.
8. Never break the live VPN bot — new work gets its own token, directory, service, DB, and capped resources.
9. Ask before state-changing work — read-only checks are free to run; anything destructive needs approval (except the standing freeze-recovery order).
10. Name the object in every closing line — a bare "ပိတ်လိုက်ပြီ" makes him ask what was closed.