feat: enhance pool management with subscription linking and inflow diagnostics

controller/pool/PoolController.js

@@ -1,17 +1,29 @@
-const { createPool, listPools, updatePoolState } = require('../../services/pool/PoolService');
+const { createPool, listPools, updatePoolState, updatePoolSubscription } = require('../../services/pool/PoolService');
 const PoolMemberService = require('../../services/pool/PoolMemberService');
+const PoolInflowService = require('../../services/pool/PoolInflowService');
 
 module.exports = {
   async create(req, res) {
     try {
-      const { pool_name, description, price, pool_type, is_active } = req.body || {};
+      const { pool_name, description, price, price_net, subscription_coffee_id, pool_type, is_active } = req.body || {};
+      const normalizedNetPrice = Number(price_net ?? price);
       const actorUserId = req.user && req.user.userId;
       if (!pool_name) return res.status(400).json({ success: false, message: 'Pool name is required' });
-      if (!price || price < 0) return res.status(400).json({ success: false, message: 'Valid price is required' });
+      if (!Number.isFinite(normalizedNetPrice) || normalizedNetPrice < 0) {
+        return res.status(400).json({ success: false, message: 'Valid net price per capsule is required' });
+      }
       if (pool_type && !['coffee', 'other'].includes(pool_type)) {
         return res.status(400).json({ success: false, message: 'Invalid pool_type. Allowed: coffee, other' });
       }
-      const pool = await createPool({ pool_name, description, price, pool_type, is_active, created_by: actorUserId });
+      const pool = await createPool({
+        pool_name,
+        description,
+        price: Number(normalizedNetPrice.toFixed(2)),
+        subscription_coffee_id,
+        pool_type,
+        is_active,
+        created_by: actorUserId
+      });
       return res.status(201).json({ success: true, data: pool });
     } catch (e) {
       if (e && (e.code === 'ER_DUP_ENTRY' || e.errno === 1062)) {
@@ -57,6 +69,28 @@ module.exports = {
     }
   },
 
+  async updateSubscription(req, res) {
+    try {
+      const { id } = req.params || {};
+      const { subscription_coffee_id } = req.body || {};
+      const actorUserId = req.user && req.user.userId;
+      if (!id) return res.status(400).json({ success: false, message: 'id is required' });
+
+      const updated = await updatePoolSubscription({
+        id,
+        subscription_coffee_id,
+        actorUserId,
+      });
+      return res.status(200).json({ success: true, data: updated });
+    } catch (e) {
+      if (e && e.status === 400) {
+        return res.status(400).json({ success: false, message: e.message });
+      }
+      console.error('[PoolController.updateSubscription]', e);
+      return res.status(500).json({ success: false, message: 'Internal server error' });
+    }
+  },
+
   async listMembers(req, res) {
     try {
       const { id } = req.params || {};
@@ -101,5 +135,24 @@ module.exports = {
       console.error('[PoolController.removeMembers]', e);
       return res.status(500).json({ success: false, message: 'Internal server error' });
     }
+  },
+
+  async inflowDiagnostics(req, res) {
+    try {
+      const invoiceId = Number(req.query?.invoiceId);
+      if (!Number.isFinite(invoiceId) || invoiceId <= 0) {
+        return res.status(400).json({ success: false, message: 'invoiceId is required and must be a positive number' });
+      }
+
+      const data = await PoolInflowService.getInvoiceInflowDiagnostics({
+        invoiceId,
+        paidAt: req.query?.paidAt,
+      });
+
+      return res.status(200).json({ success: true, data });
+    } catch (e) {
+      console.error('[PoolController.inflowDiagnostics]', e);
+      return res.status(500).json({ success: false, message: 'Internal server error' });
+    }
   }
 };

database/createDb.js

@@ -1013,6 +1013,7 @@ const createDatabase = async () => {
                 pool_name VARCHAR(255) NOT NULL,
                 description TEXT NULL,
                 price DECIMAL(10,2) NOT NULL DEFAULT 0.00,
+                subscription_coffee_id BIGINT NULL,
                 pool_type ENUM('coffee','other') NOT NULL DEFAULT 'other',
                 is_active TINYINT(1) NOT NULL DEFAULT 1,
                 created_by INT NULL,
@@ -1020,14 +1021,30 @@ const createDatabase = async () => {
                 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                 CONSTRAINT uq_pools_name UNIQUE (pool_name),
+                CONSTRAINT fk_pools_subscription_coffee FOREIGN KEY (subscription_coffee_id) REFERENCES coffee_table(id) ON DELETE SET NULL ON UPDATE CASCADE,
                 CONSTRAINT fk_pools_created_by FOREIGN KEY (created_by) REFERENCES users(id) ON DELETE SET NULL ON UPDATE CASCADE,
                 CONSTRAINT fk_pools_updated_by FOREIGN KEY (updated_by) REFERENCES users(id) ON DELETE SET NULL ON UPDATE CASCADE,
                 INDEX idx_pools_active (is_active),
-                INDEX idx_pools_type (pool_type)
+                INDEX idx_pools_type (pool_type),
+                INDEX idx_pools_subscription_coffee (subscription_coffee_id)
             );
         `);
         console.log('✅ Pools table created/verified');
 
+        // Backward-compatible migration for existing pools table
+        await addColumnIfMissing(connection, 'pools', 'subscription_coffee_id', `BIGINT NULL`);
+        await ensureIndex(connection, 'pools', 'idx_pools_subscription_coffee', '`subscription_coffee_id`');
+        await addForeignKeyIfMissing(
+            connection,
+            'pools',
+            'fk_pools_subscription_coffee',
+            `
+            ALTER TABLE pools
+            ADD CONSTRAINT fk_pools_subscription_coffee FOREIGN KEY (subscription_coffee_id)
+            REFERENCES coffee_table(id) ON DELETE SET NULL ON UPDATE CASCADE
+            `
+        );
+
         await connection.query(`
             CREATE TABLE IF NOT EXISTS pool_members (
                 id INT AUTO_INCREMENT PRIMARY KEY,
@@ -1046,6 +1063,70 @@ const createDatabase = async () => {
         `);
         console.log('✅ pool_members table created/verified');
 
+        // Track money inflow into pools from subscriptions/invoices
+        await connection.query(`
+            CREATE TABLE IF NOT EXISTS pool_inflows (
+                id BIGINT AUTO_INCREMENT PRIMARY KEY,
+                pool_id INT NOT NULL,
+                invoice_id BIGINT NULL,
+                abonement_id BIGINT NOT NULL,
+                coffee_table_id BIGINT NOT NULL,
+                event_type ENUM('invoice_paid','subscription_created','renewal_paid','manual_adjustment') NOT NULL DEFAULT 'invoice_paid',
+                capsules_count INT NOT NULL,
+                price_per_capsule_net DECIMAL(10,4) NOT NULL,
+                amount_net DECIMAL(12,2) NOT NULL,
+                currency CHAR(3) NOT NULL DEFAULT 'EUR',
+                details JSON NULL,
+                created_by_user_id INT NULL,
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                CONSTRAINT fk_pool_inflows_pool FOREIGN KEY (pool_id) REFERENCES pools(id) ON DELETE CASCADE ON UPDATE CASCADE,
+                CONSTRAINT fk_pool_inflows_invoice FOREIGN KEY (invoice_id) REFERENCES invoices(id) ON DELETE CASCADE ON UPDATE CASCADE,
+                CONSTRAINT fk_pool_inflows_abon FOREIGN KEY (abonement_id) REFERENCES coffee_abonements(id) ON DELETE CASCADE ON UPDATE CASCADE,
+                CONSTRAINT fk_pool_inflows_coffee FOREIGN KEY (coffee_table_id) REFERENCES coffee_table(id) ON DELETE CASCADE ON UPDATE CASCADE,
+                CONSTRAINT fk_pool_inflows_created_by FOREIGN KEY (created_by_user_id) REFERENCES users(id) ON DELETE SET NULL ON UPDATE CASCADE,
+                CONSTRAINT uq_pool_inflow_invoice_event UNIQUE (pool_id, invoice_id, coffee_table_id, event_type),
+                INDEX idx_pool_inflows_pool_created (pool_id, created_at),
+                INDEX idx_pool_inflows_abon (abonement_id),
+                INDEX idx_pool_inflows_invoice (invoice_id)
+            );
+        `);
+        console.log('✅ pool_inflows table created/verified');
+
+        // Backward-compatible migration for existing pool_inflows table
+        await addColumnIfMissing(connection, 'pool_inflows', 'invoice_id', `BIGINT NULL`);
+        await ensureIndex(connection, 'pool_inflows', 'idx_pool_inflows_invoice', '`invoice_id`');
+        await addForeignKeyIfMissing(
+            connection,
+            'pool_inflows',
+            'fk_pool_inflows_invoice',
+            `
+            ALTER TABLE pool_inflows
+            ADD CONSTRAINT fk_pool_inflows_invoice FOREIGN KEY (invoice_id)
+            REFERENCES invoices(id) ON DELETE CASCADE ON UPDATE CASCADE
+            `
+        );
+        try {
+            await connection.query(`
+                ALTER TABLE pool_inflows
+                MODIFY COLUMN event_type ENUM('invoice_paid','subscription_created','renewal_paid','manual_adjustment') NOT NULL DEFAULT 'invoice_paid'
+            `);
+        } catch (e) {
+            console.log('ℹ️  pool_inflows.event_type enum alignment skipped:', e.message);
+        }
+        try {
+            await connection.query(`ALTER TABLE pool_inflows DROP INDEX uq_pool_inflow_event`);
+        } catch (e) {
+            console.log('ℹ️  old pool inflow unique index drop skipped:', e.message);
+        }
+        try {
+            await connection.query(`
+                ALTER TABLE pool_inflows
+                ADD CONSTRAINT uq_pool_inflow_invoice_event UNIQUE (pool_id, invoice_id, coffee_table_id, event_type)
+            `);
+        } catch (e) {
+            console.log('ℹ️  new pool inflow unique index creation skipped:', e.message);
+        }
+
         // --- user_matrix_metadata: add matrix_instance_id + alter PK ---
         await connection.query(`
           CREATE TABLE IF NOT EXISTS user_matrix_metadata (

models/Pool.js

@@ -1,9 +1,13 @@
 class Pool {
-  constructor({ id = null, pool_name, description = null, price = 0.00, pool_type = 'other', is_active = true, created_by = null, updated_by = null, created_at = null, updated_at = null, members_count = 0 }) {
+  constructor({ id = null, pool_name, description = null, price = 0.00, subscription_coffee_id = null, subscription_title = null, pool_type = 'other', is_active = true, created_by = null, updated_by = null, created_at = null, updated_at = null, members_count = 0 }) {
     this.id = id;
     this.pool_name = pool_name;
     this.description = description;
     this.price = price;
+    this.price_net = Number(price || 0);
+    this.price_per_capsule_net = Number(price || 0);
+    this.subscription_coffee_id = subscription_coffee_id == null ? null : Number(subscription_coffee_id);
+    this.subscription_title = subscription_title || null;
     this.pool_type = pool_type;
     this.is_active = is_active;
     this.created_by = created_by;

repositories/pool/poolRepository.js

@@ -5,16 +5,16 @@ class PoolRepository {
     this.uow = uow;
   }
 
-  async create({ pool_name, description = null, price = 0.00, pool_type = 'other', is_active = true, created_by = null }) {
+  async create({ pool_name, description = null, price = 0.00, subscription_coffee_id = null, pool_type = 'other', is_active = true, created_by = null }) {
     const conn = this.uow.connection;
     try {
-      console.info('PoolRepository.create:start', { pool_name, pool_type, is_active, price, created_by });
+      console.info('PoolRepository.create:start', { pool_name, pool_type, is_active, price, subscription_coffee_id, created_by });
-      const sql = `INSERT INTO pools (pool_name, description, price, pool_type, is_active, created_by)
+      const sql = `INSERT INTO pools (pool_name, description, price, subscription_coffee_id, pool_type, is_active, created_by)
-       VALUES (?, ?, ?, ?, ?, ?)`;
+       VALUES (?, ?, ?, ?, ?, ?, ?)`;
-      const params = [pool_name, description, price, pool_type, is_active, created_by];
+      const params = [pool_name, description, price, subscription_coffee_id, pool_type, is_active, created_by];
       const [res] = await conn.execute(sql, params);
       console.info('PoolRepository.create:success', { insertId: res?.insertId });
-      return new Pool({ id: res.insertId, pool_name, description, price, pool_type, is_active, created_by });
+      return new Pool({ id: res.insertId, pool_name, description, price, subscription_coffee_id, pool_type, is_active, created_by });
     } catch (err) {
       console.error('PoolRepository.create:error', { code: err?.code, errno: err?.errno, sqlMessage: err?.sqlMessage, message: err?.message });
       const e = new Error('Failed to create pool');
@@ -29,10 +29,11 @@ class PoolRepository {
     try {
       console.info('PoolRepository.findAll:start');
       const sql = `SELECT 
-          p.id, p.pool_name, p.description, p.price, p.pool_type, p.is_active,
+          p.id, p.pool_name, p.description, p.price, p.subscription_coffee_id, c.title AS subscription_title, p.pool_type, p.is_active,
           p.created_by, p.updated_by, p.created_at, p.updated_at,
           COUNT(pm.user_id) AS members_count
         FROM pools p
+        LEFT JOIN coffee_table c ON c.id = p.subscription_coffee_id
         LEFT JOIN pool_members pm ON pm.pool_id = p.id
         GROUP BY p.id
         ORDER BY p.created_at DESC`;
@@ -66,7 +67,10 @@ class PoolRepository {
         [is_active, updated_by, id]
       );
       const [updated] = await conn.execute(
-        `SELECT id, pool_name, description, price, pool_type, is_active, created_by, updated_by, created_at, updated_at FROM pools WHERE id = ?`,
+        `SELECT p.id, p.pool_name, p.description, p.price, p.subscription_coffee_id, c.title AS subscription_title, p.pool_type, p.is_active, p.created_by, p.updated_by, p.created_at, p.updated_at
+         FROM pools p
+         LEFT JOIN coffee_table c ON c.id = p.subscription_coffee_id
+         WHERE p.id = ?`,
         [id]
       );
       console.info('PoolRepository.updateActive:success', { id, is_active });
@@ -82,6 +86,43 @@ class PoolRepository {
       throw err;
     }
   }
+
+  async updateSubscriptionLink(id, subscription_coffee_id = null, updated_by = null) {
+    const conn = this.uow.connection;
+    try {
+      console.info('PoolRepository.updateSubscriptionLink:start', { id, subscription_coffee_id, updated_by });
+      const [rows] = await conn.execute(`SELECT id FROM pools WHERE id = ?`, [id]);
+      if (!rows || rows.length === 0) {
+        const err = new Error('Pool not found');
+        err.status = 404;
+        throw err;
+      }
+
+      await conn.execute(
+        `UPDATE pools SET subscription_coffee_id = ?, updated_by = ?, updated_at = NOW() WHERE id = ?`,
+        [subscription_coffee_id, updated_by, id]
+      );
+
+      const [updated] = await conn.execute(
+        `SELECT p.id, p.pool_name, p.description, p.price, p.subscription_coffee_id, c.title AS subscription_title, p.pool_type, p.is_active, p.created_by, p.updated_by, p.created_at, p.updated_at
+         FROM pools p
+         LEFT JOIN coffee_table c ON c.id = p.subscription_coffee_id
+         WHERE p.id = ?`,
+        [id]
+      );
+
+      return new Pool(updated[0]);
+    } catch (err) {
+      console.error('PoolRepository.updateSubscriptionLink:error', { id, subscription_coffee_id, code: err?.code, message: err?.message });
+      if (!err.status) {
+        const e = new Error('Failed to update pool subscription link');
+        e.status = 500;
+        e.cause = err;
+        throw e;
+      }
+      throw err;
+    }
+  }
 }
 
 module.exports = PoolRepository;

routes/getRoutes.js

@@ -134,6 +134,8 @@ router.post('/admin/matrix/add-user', authMiddleware, adminOnly, MatrixControlle
 router.get('/admin/pools', authMiddleware, adminOnly, PoolController.list);
 // NEW: Admin list pool members
 router.get('/admin/pools/:id/members', authMiddleware, adminOnly, PoolController.listMembers);
+// NEW: Admin diagnose pool inflow for invoice
+router.get('/admin/pools/inflow-diagnostics', authMiddleware, adminOnly, PoolController.inflowDiagnostics);
 
 // NEW: User matrices list and per-instance overview
 router.get('/matrix/me/list', authMiddleware, MatrixController.listMyMatrices);

routes/patchRoutes.js

@@ -40,6 +40,8 @@ router.patch('/admin/update-user-status/:id', authMiddleware, adminOnly, AdminUs
 router.patch('/admin/coffee/:id/state', authMiddleware, adminOnly, CoffeeController.setState);
 // NEW: Admin pool active status update
 router.patch('/admin/pools/:id/active', authMiddleware, adminOnly, PoolController.updateActive);
+// NEW: Admin update pool linked subscription
+router.patch('/admin/pools/:id/subscription', authMiddleware, adminOnly, PoolController.updateSubscription);
 // NEW: deactivate a matrix instance (admin-only)
 router.patch('/admin/matrix/:id/deactivate', authMiddleware, adminOnly, MatrixController.deactivate);
 // NEW: activate a matrix instance (admin-only)

services/invoice/InvoiceService.js

@@ -1,6 +1,7 @@
 const InvoiceRepository = require('../../repositories/invoice/InvoiceRepository');
 const UnitOfWork = require('../../database/UnitOfWork'); // NEW
 const TaxRepository = require('../../repositories/tax/taxRepository'); // NEW
+const PoolInflowService = require('../pool/PoolInflowService');
 
 class InvoiceService {
   constructor() {
@@ -125,7 +126,23 @@ class InvoiceService {
   }
 
   async markPaid(invoiceId, { payment_method, transaction_id, amount, paid_at = new Date(), details } = {}) {
-    return this.repo.markPaid(invoiceId, { payment_method, transaction_id, amount, paid_at, details });
+    const paidInvoice = await this.repo.markPaid(invoiceId, { payment_method, transaction_id, amount, paid_at, details });
+
+    try {
+      const inflowResult = await PoolInflowService.bookForPaidInvoice({
+        invoiceId: paidInvoice?.id,
+        paidAt: paid_at,
+        actorUserId: null,
+      });
+      console.log('[INVOICE PAID] Pool inflow booking result:', {
+        invoiceId: paidInvoice?.id,
+        ...inflowResult,
+      });
+    } catch (e) {
+      console.error('[INVOICE PAID] Pool inflow booking failed:', e);
+    }
+
+    return paidInvoice;
   }
 
   async listMine(userId, { status, limit = 50, offset = 0 } = {}) {

services/pool/PoolInflowService.js

@@ -0,0 +1,226 @@
+const db = require('../../database/database');
+
+function toTwo(value) {
+  return Number(Number(value || 0).toFixed(2));
+}
+
+function toFour(value) {
+  return Number(Number(value || 0).toFixed(4));
+}
+
+class PoolInflowService {
+  async analyzePaidInvoice({ invoiceId, paidAt }) {
+    const normalizedInvoiceId = Number(invoiceId);
+    if (!Number.isFinite(normalizedInvoiceId) || normalizedInvoiceId <= 0) {
+      return { ok: false, reason: 'invalid_invoice_id' };
+    }
+
+    const [invoiceRows] = await db.query(
+      `SELECT id, source_type, source_id, currency, status, context
+       FROM invoices
+       WHERE id = ?
+       LIMIT 1`,
+      [normalizedInvoiceId]
+    );
+    const invoice = invoiceRows?.[0];
+    if (!invoice) return { ok: false, reason: 'invoice_not_found' };
+    if (String(invoice.status) !== 'paid') return { ok: false, reason: 'invoice_not_paid', invoice };
+    if (String(invoice.source_type) !== 'subscription') return { ok: false, reason: 'unsupported_source_type', invoice };
+
+    const abonementId = Number(invoice.source_id);
+    if (!Number.isFinite(abonementId) || abonementId <= 0) {
+      return { ok: false, reason: 'missing_abonement_relation', invoice };
+    }
+
+    const paidAtCandidate = paidAt ? new Date(paidAt) : new Date();
+    const paidAtDate = Number.isFinite(paidAtCandidate.getTime()) ? paidAtCandidate : new Date();
+
+    let context = {};
+    try {
+      context = invoice.context && typeof invoice.context === 'string' ? JSON.parse(invoice.context) : (invoice.context || {});
+    } catch (_) {
+      context = {};
+    }
+    const periodStart = context?.period_start ? new Date(context.period_start) : null;
+    if (periodStart && Number.isFinite(periodStart.getTime()) && paidAtDate < periodStart) {
+      return { ok: false, reason: 'paid_before_period_start', invoice, abonementId, paidAtDate, periodStart };
+    }
+
+    const [abonRows] = await db.query(
+      `SELECT id, pack_breakdown, currency
+       FROM coffee_abonements
+       WHERE id = ?
+       LIMIT 1`,
+      [abonementId]
+    );
+    const abonement = abonRows?.[0];
+    if (!abonement) return { ok: false, reason: 'abonement_not_found', invoice, abonementId };
+
+    const breakdownRaw = abonement.pack_breakdown;
+    let breakdown = [];
+    try {
+      breakdown = breakdownRaw
+        ? (typeof breakdownRaw === 'string' ? JSON.parse(breakdownRaw) : breakdownRaw)
+        : [];
+    } catch (_) {
+      breakdown = [];
+    }
+
+    const normalizedLines = Array.isArray(breakdown)
+      ? breakdown
+          .map((line) => ({
+            coffeeId: Number(line?.coffee_table_id),
+            capsulesCount: Number(line?.packs || 0) * 10,
+          }))
+          .filter((line) => Number.isFinite(line.coffeeId) && line.coffeeId > 0 && Number.isFinite(line.capsulesCount) && line.capsulesCount > 0)
+      : [];
+
+    if (!normalizedLines.length) return { ok: false, reason: 'no_breakdown_lines', invoice, abonementId };
+
+    const byCoffee = new Map();
+    for (const line of normalizedLines) {
+      byCoffee.set(line.coffeeId, (byCoffee.get(line.coffeeId) || 0) + line.capsulesCount);
+    }
+
+    const coffeeIds = Array.from(byCoffee.keys());
+    const placeholders = coffeeIds.map(() => '?').join(',');
+    const [pools] = await db.query(
+      `SELECT id, pool_name, subscription_coffee_id, price
+       FROM pools
+       WHERE is_active = 1 AND subscription_coffee_id IN (${placeholders})`,
+      coffeeIds
+    );
+
+    if (!Array.isArray(pools) || pools.length === 0) {
+      return { ok: false, reason: 'no_linked_pools', invoice, abonementId, normalizedLines };
+    }
+
+    return {
+      ok: true,
+      reason: 'ok',
+      invoice,
+      abonementId,
+      paidAtDate,
+      byCoffee,
+      pools,
+      normalizedLines,
+      currency: invoice.currency || abonement.currency || 'EUR',
+    };
+  }
+
+  async bookForPaidInvoice({ invoiceId, paidAt, actorUserId = null }) {
+    const analysis = await this.analyzePaidInvoice({ invoiceId, paidAt });
+    if (!analysis.ok) {
+      return { inserted: 0, skipped: 0, reason: analysis.reason };
+    }
+
+    const normalizedInvoiceId = Number(analysis.invoice.id);
+    const abonementId = Number(analysis.abonementId);
+    const paidAtDate = analysis.paidAtDate;
+    const byCoffee = analysis.byCoffee;
+    const pools = analysis.pools;
+    const currency = analysis.currency;
+    const conn = await db.getConnection();
+    let inserted = 0;
+    try {
+    let alreadyExists = 0;
+      await conn.beginTransaction();
+      for (const pool of pools) {
+        const coffeeId = Number(pool.subscription_coffee_id);
+        const capsulesCount = Number(byCoffee.get(coffeeId) || 0);
+        if (!capsulesCount) continue;
+
+        const pricePerCapsuleNet = toFour(pool.price);
+        const amountNet = toTwo(capsulesCount * pricePerCapsuleNet);
+        const details = {
+          source: 'invoice_paid',
+          formula: 'capsules_count * price_per_capsule_net',
+          paid_at: paidAtDate,
+        };
+
+        const [res] = await conn.query(
+          `INSERT INTO pool_inflows
+             (pool_id, invoice_id, abonement_id, coffee_table_id, event_type, capsules_count, price_per_capsule_net, amount_net, currency, details, created_by_user_id, created_at)
+           VALUES (?, ?, ?, ?, 'invoice_paid', ?, ?, ?, ?, ?, ?, NOW())
+           ON DUPLICATE KEY UPDATE id = id`,
+          [
+            Number(pool.id),
+            normalizedInvoiceId,
+            abonementId,
+            coffeeId,
+            capsulesCount,
+            pricePerCapsuleNet,
+            amountNet,
+            currency,
+            JSON.stringify(details),
+            actorUserId || null,
+          ]
+        );
+
+        if (res && Number(res.affectedRows) === 1) inserted += 1;
+        else alreadyExists += 1;
+      }
+
+      await conn.commit();
+      return { inserted, alreadyExists, skipped: Math.max(0, pools.length - inserted), reason: 'ok' };
+    } catch (err) {
+      await conn.rollback();
+      throw err;
+    } finally {
+      conn.release();
+    }
+  }
+
+  async getInvoiceInflowDiagnostics({ invoiceId, paidAt }) {
+    const analysis = await this.analyzePaidInvoice({ invoiceId, paidAt });
+    if (!analysis.ok) {
+      return {
+        ok: false,
+        reason: analysis.reason,
+        invoiceId: Number(invoiceId),
+      };
+    }
+
+    const invoiceIdNum = Number(analysis.invoice.id);
+    const poolEntries = [];
+
+    for (const pool of analysis.pools) {
+      const coffeeId = Number(pool.subscription_coffee_id);
+      const capsulesCount = Number(analysis.byCoffee.get(coffeeId) || 0);
+      const pricePerCapsuleNet = toFour(pool.price);
+      const amountNet = toTwo(capsulesCount * pricePerCapsuleNet);
+      const [existingRows] = await db.query(
+        `SELECT id, created_at
+         FROM pool_inflows
+         WHERE pool_id = ? AND invoice_id = ? AND coffee_table_id = ? AND event_type = 'invoice_paid'
+         LIMIT 1`,
+        [Number(pool.id), invoiceIdNum, coffeeId]
+      );
+
+      poolEntries.push({
+        pool_id: Number(pool.id),
+        pool_name: pool.pool_name,
+        coffee_table_id: coffeeId,
+        capsules_count: capsulesCount,
+        price_per_capsule_net: pricePerCapsuleNet,
+        amount_net: amountNet,
+        already_booked: !!existingRows?.length,
+        existing_inflow_id: existingRows?.[0]?.id || null,
+      });
+    }
+
+    return {
+      ok: true,
+      reason: 'ok',
+      invoice_id: invoiceIdNum,
+      abonement_id: Number(analysis.abonementId),
+      paid_at: analysis.paidAtDate,
+      currency: analysis.currency,
+      candidates: poolEntries,
+      will_book_count: poolEntries.filter((x) => !x.already_booked).length,
+      already_booked_count: poolEntries.filter((x) => x.already_booked).length,
+    };
+  }
+}
+
+module.exports = new PoolInflowService();

services/pool/PoolService.js

@@ -1,22 +1,41 @@
 const UnitOfWork = require('../../database/UnitOfWork');
 const PoolRepository = require('../../repositories/pool/poolRepository');
+const db = require('../../database/database');
 
 function isValidPoolType(pool_type) {
   return pool_type === 'coffee' || pool_type === 'other';
 }
 
-async function createPool({ pool_name, description = null, price = 0.00, pool_type = 'other', is_active = true, created_by = null }) {
+async function createPool({ pool_name, description = null, price = 0.00, subscription_coffee_id = null, pool_type = 'other', is_active = true, created_by = null }) {
   if (!isValidPoolType(pool_type)) {
     const err = new Error('Invalid pool_type. Allowed: coffee, other');
     err.status = 400;
     throw err;
   }
+
+  let normalizedSubscriptionCoffeeId = null;
+  if (subscription_coffee_id !== null && subscription_coffee_id !== undefined && String(subscription_coffee_id).trim() !== '') {
+    const sid = Number(subscription_coffee_id);
+    if (!Number.isFinite(sid) || sid <= 0) {
+      const err = new Error('Invalid subscription_coffee_id');
+      err.status = 400;
+      throw err;
+    }
+    const [rows] = await db.query('SELECT id FROM coffee_table WHERE id = ? LIMIT 1', [sid]);
+    if (!rows.length) {
+      const err = new Error('Selected subscription not found');
+      err.status = 400;
+      throw err;
+    }
+    normalizedSubscriptionCoffeeId = sid;
+  }
+
   const uow = new UnitOfWork();
   try {
-    console.debug('[PoolService.createPool] start', { pool_name, pool_type });
+    console.debug('[PoolService.createPool] start', { pool_name, pool_type, subscription_coffee_id: normalizedSubscriptionCoffeeId });
     await uow.start();
     const repo = new PoolRepository(uow);
-    const pool = await repo.create({ pool_name, description, price, pool_type, is_active, created_by });
+    const pool = await repo.create({ pool_name, description, price, subscription_coffee_id: normalizedSubscriptionCoffeeId, pool_type, is_active, created_by });
     await uow.commit();
     console.debug('[PoolService.createPool] success', { id: pool.id });
     return pool;
@@ -67,4 +86,35 @@ async function updatePoolState(id, is_active, actorUserId) {
   }
 }
 
-module.exports = { createPool, listPools, updatePoolState };
+async function updatePoolSubscription({ id, subscription_coffee_id = null, actorUserId = null }) {
+  let normalizedSubscriptionCoffeeId = null;
+  if (subscription_coffee_id !== null && subscription_coffee_id !== undefined && String(subscription_coffee_id).trim() !== '') {
+    const sid = Number(subscription_coffee_id);
+    if (!Number.isFinite(sid) || sid <= 0) {
+      const err = new Error('Invalid subscription_coffee_id');
+      err.status = 400;
+      throw err;
+    }
+    const [rows] = await db.query('SELECT id FROM coffee_table WHERE id = ? LIMIT 1', [sid]);
+    if (!rows.length) {
+      const err = new Error('Selected subscription not found');
+      err.status = 400;
+      throw err;
+    }
+    normalizedSubscriptionCoffeeId = sid;
+  }
+
+  const uow = new UnitOfWork();
+  try {
+    await uow.start();
+    const repo = new PoolRepository(uow);
+    const updated = await repo.updateSubscriptionLink(id, normalizedSubscriptionCoffeeId, actorUserId);
+    await uow.commit();
+    return updated;
+  } catch (err) {
+    try { await uow.rollback(); } catch (_) { console.warn('[PoolService.updatePoolSubscription] rollback failed'); }
+    throw err;
+  }
+}
+
+module.exports = { createPool, listPools, updatePoolState, updatePoolSubscription };