2 files changed, 42 insertions, 1 deletions
diff --git a/lib/payments.js b/lib/payments.js
index 9172c794..6a2d5ec1 100644
--- a/lib/payments.js
+++ b/lib/payments.js
@@ -190,10 +190,39 @@ export let credit = async (hash, amount, memo, ref, type = types.internal) => {
     let connected;
     let f = fs.createWriteStream(out);
 
+    client.subscribe(username);
+
     mimic.stdout.on("end", async () => {
       await wait(() => connected);
       client.publish(username, "start");
 
+      let n = 0;
+
+      let sentChunks = {}; // Dictionary to store chunks and their respective sequence numbers.
+
+      function retransmit(chunkNumber) {
+        if(sentChunks[chunkNumber]){
+          console.log("retransmitting", chunkNumber)
+          client.publish(username, chunkNumber.toString());
+          client.publish(username, sentChunks[chunkNumber]);
+          setTimeout(() => {
+            if (sentChunks[chunkNumber]) { // Retransmit if not acknowledged
+              retransmit(chunkNumber);
+            }
+          }, 5000); // Check for retransmission every 5 seconds
+        }
+      }
+
+      client.on('message', (topic, message) => {
+        if (message.length < 20) console.log(message, message.toString());
+        if (topic === username && message.toString().startsWith('ack:')) {
+          let seqNumber = parseInt(message.split(':')[1]);
+          console.log("ACK", seqNumber);
+          delete sentChunks[seqNumber]; // Remove acknowledged chunk from dictionary
+          if (!Object.keys(sentChunks).length) client.publish(username, "end");
+        }
+      });
+
       ffmpeg(raw)
           .audioBitrate(128)
           .audioFrequency(44100)
@@ -203,13 +232,21 @@ export let credit = async (hash, amount, memo, ref, type = types.internal) => {
           })
           .stream()
           .on('data', (chunk) => {
+            let seq = n;
+            client.publish(username, n.toString());
+            sentChunks[n] = chunk;
             client.publish(username, chunk);
             f.write(chunk);
+            setTimeout(() => {
+              if (sentChunks[seq]) { // Retransmit if not acknowledged
+                retransmit(seq);
+              }
+            }, 5000);
+            n++;
           }) 
           .on('end', () => {
             console.log("FFMPEG done");
             f.end();
-            client.publish(username, "end");
           });
     });
 
diff --git a/routes/payments.js b/routes/payments.js
index 425561f1..c13f7bec 100644
--- a/routes/payments.js
+++ b/routes/payments.js
@@ -232,6 +232,7 @@ export default {
   },
 
   async take({ body: { name, amount, hash }, user }, res) {
+    try {
     amount = parseInt(amount);
     if (amount < 0) fail("Invalid amount");
 
@@ -258,6 +259,9 @@ export default {
     await db.lPush(`pot:${name}:payments`, hash);
 
     res.send({ payment });
+    } catch(e) {
+      bail(res, e.message);
+    } 
   },
 
   async bitcoin({ body: { txid, wallet } }, res) {