openvibers/openvibe.live Commits
PowerChat leaderboards: feed channel-point EARNS, not just redemptions
Changes
@@@ -989,6 +989,8 @@ async function start() {try { require('./media-proxy/asset-sync').start(); } catch (e) { console.warn('[AssetSync] not started:', e.message); }// Feed connected streamers' viewer counts into PowerChat overlays (platform mode).try { require('./integrations/powerchat-platform').startViewerCountSweeper(); } catch (e) { console.warn('[PowerChat] viewer sweeper not started:', e.message); }++ // Batched channel-point earns → PowerChat leaderboard feed.++ try { require('./integrations/powerchat-platform').startCurrencyEarnFlusher(); } catch (e) { console.warn('[PowerChat] earn flusher not started:', e.message); }// PowerChat: prune the webhook-dedupe log daily so it can't grow unbounded.try {const _pcClean = () => { try { db.cleanupPowerchatDeliveries(3); } catch { /* */ } };@@@ -249,11 +249,12 @@ function startViewerCountSweeper() {}// ── currency:write ───────────────────────────────────────────++// Returns true when PowerChat accepted the event (the earn flusher re-buffers on false).async function sendCurrencyRedemption(streamerUserId, { amount, redeemerName, rewardName, message, externalId } = {}) {const conn = _connFor(streamerUserId, 'currency:write');-- if (!conn) return;++ if (!conn) return false;const amt = Math.round(Number(amount) || 0);-- if (amt < 1) return; // schema requires amount 1-1000000000++ if (amt < 1) return false; // schema requires amount 1-1000000000try {await oauth.apiRequest(streamerUserId, {method: 'POST', path: '/currency-events',@@@ -270,6 +271,7 @@ async function sendCurrencyRedemption(streamerUserId, { amount, redeemerName, reoccurredAt: new Date().toISOString(),},});++ return true;} catch (e) {_noteApiError(streamerUserId, 'currency:write', e);// "Unknown currency" means the key isn't DECLARED on the app in the PowerChat@@@ -277,9 +279,62 @@ async function sendCurrencyRedemption(streamerUserId, { amount, redeemerName, reif (e && e.status === 400 && /unknown currency/i.test(e.message || '')) {console.warn(`[PowerChat] currency "${CURRENCY_KEY}" is not declared on the app — declare it in the PowerChat Developer dashboard`);}++ return false;}}++// ── Channel-point EARNS → PowerChat leaderboard feed ─────────++// Redemptions alone can't populate a leaderboard (they're rare); the board lives on++// the steady drip of watch/chat/follow point earning. Every award is buffered per++// (streamer, viewer) and flushed as ONE summed currency event per viewer every few++// minutes — that keeps a busy channel inside the ~60/min currency rate limit and++// avoids machine-gunning the overlay with micro-events.++const EARN_FLUSH_MS = 5 * 60 * 1000;++const EARN_MAX_PER_FLUSH = 40; // stay well under the rate limit per flush pass++const EARN_MAX_VIEWERS = 500; // per-streamer buffer cap (drop past this, don't grow)++const _earnBuf = new Map(); // streamerId → Map(viewerId → summed amount)++function queueCurrencyEarn(streamerUserId, viewerUserId, amount) {++ const amt = Math.round(Number(amount) || 0);++ if (amt < 1 || !streamerUserId || !viewerUserId) return;++ // Only buffer for streamers with a live currency:write connection — otherwise the++ // buffer would grow forever for channels that never flush.++ if (!_connFor(streamerUserId, 'currency:write')) return;++ let m = _earnBuf.get(streamerUserId);++ if (!m) { m = new Map(); _earnBuf.set(streamerUserId, m); }++ if (!m.has(viewerUserId) && m.size >= EARN_MAX_VIEWERS) return;++ m.set(viewerUserId, (m.get(viewerUserId) || 0) + amt);++}++async function _flushCurrencyEarns() {++ let sent = 0;++ for (const [streamerId, m] of _earnBuf) {++ if (!m.size) { _earnBuf.delete(streamerId); continue; }++ for (const [viewerId, amount] of m) {++ if (sent >= EARN_MAX_PER_FLUSH) return; // leave the rest buffered for next pass++ m.delete(viewerId);++ sent++;++ let name = null;++ try { const u = db.getUserById(viewerId); name = u ? (u.display_name || u.username) : null; } catch { /* */ }++ const ok = await sendCurrencyRedemption(streamerId, {++ amount,++ redeemerName: name || `viewer ${viewerId}`,++ rewardName: 'Points earned',++ externalId: `earn:${streamerId}:${viewerId}:${Date.now()}`,++ });++ // Transient failure → put the amount back so the points aren't lost.++ if (!ok && _connFor(streamerId, 'currency:write')) {++ m.set(viewerId, (m.get(viewerId) || 0) + amount);++ }++ }++ }++}++let _earnTimer = null;++function startCurrencyEarnFlusher() {++ if (_earnTimer) return;++ _earnTimer = setInterval(() => { _flushCurrencyEarns().catch(() => { }); }, EARN_FLUSH_MS);++ if (_earnTimer.unref) _earnTimer.unref();++ console.log('[PowerChat] channel-point earn flusher started (leaderboard feed)');++}++// ── tips:write ───────────────────────────────────────────────// Forward an on-site Vibes donation as a MONETARY tip. amount is in Vibes UNITS;// PowerChat converts server-side via the declared unitsPerUsd (100 Vibes = $1) and the@@@ -333,5 +388,5 @@ module.exports = {CURRENCY_KEY, TIP_CURRENCY_KEY,forwardChat, forwardFollow, forwardSubscription, forwardTip,sendViewCount, startViewerCountSweeper,-- sendCurrencyRedemption, sendCustomAlert,++ sendCurrencyRedemption, queueCurrencyEarn, startCurrencyEarnFlusher, sendCustomAlert,};@@@ -31,6 +31,12 @@ const COINS = {STREAK_THRESHOLD_MIN: 60, // minutes before streak kicks in};++// Mirror every channel-point award into the streamer's PowerChat leaderboard feed++// (batched there; a no-op unless the streamer connected PowerChat with currency:write).++function _feedPowerchat(streamerId, userId, coins) {++ try { require('../integrations/powerchat-platform').queueCurrencyEarn(streamerId, userId, coins); } catch { /* non-critical */ }++}++// In-memory cooldown tracker (userId → lastChatCoinTime)const chatCooldowns = new Map();// Bonus-game claim throttle ("userId:streamerId" → last claim ms)@@@ -69,6 +75,7 @@ class OpenCoins {}const total = db.addChannelPoints(userId, streamerId, coins);++ _feedPowerchat(streamerId, userId, coins);db.createCoinTransaction({user_id: userId,stream_id: streamId,@@@ -104,6 +111,7 @@ class OpenCoins {chatCooldowns.set(userId, now);const total = db.addChannelPoints(userId, streamerId, COINS.CHAT_BONUS);++ _feedPowerchat(streamerId, userId, COINS.CHAT_BONUS);db.createCoinTransaction({user_id: userId,stream_id: streamId,@@@ -132,6 +140,7 @@ class OpenCoins {if (!streamerId || streamerId === userId) return null;const total = db.addChannelPoints(userId, streamerId, COINS.FOLLOW_BONUS);++ _feedPowerchat(streamerId, userId, COINS.FOLLOW_BONUS);db.createCoinTransaction({user_id: userId,stream_id: null,@@@ -248,6 +257,7 @@ class OpenCoins {bonusClaims.set(key, now);const amount = Math.max(1, (cfg.watch_amount || 10) * 3);const total = db.addChannelPoints(userId, streamerId, amount);++ _feedPowerchat(streamerId, userId, amount);db.createCoinTransaction({ user_id: userId, stream_id: streamId, amount, type: 'watch', message: 'Bonus game' });return { coins: amount, total, streamerId };}
@@ -989,6 +989,8 @@ async function start() {
@@ -989,6 +989,8 @@ async function start() {
try { require('./media-proxy/asset-sync').start(); } catch (e) { console.warn('[AssetSync] not started:', e.message); }
try { require('./media-proxy/asset-sync').start(); } catch (e) { console.warn('[AssetSync] not started:', e.message); }
// Feed connected streamers' viewer counts into PowerChat overlays (platform mode).
// Feed connected streamers' viewer counts into PowerChat overlays (platform mode).
try { require('./integrations/powerchat-platform').startViewerCountSweeper(); } catch (e) { console.warn('[PowerChat] viewer sweeper not started:', e.message); }
try { require('./integrations/powerchat-platform').startViewerCountSweeper(); } catch (e) { console.warn('[PowerChat] viewer sweeper not started:', e.message); }
+ // Batched channel-point earns → PowerChat leaderboard feed.
+ try { require('./integrations/powerchat-platform').startCurrencyEarnFlusher(); } catch (e) { console.warn('[PowerChat] earn flusher not started:', e.message); }
// PowerChat: prune the webhook-dedupe log daily so it can't grow unbounded.
// PowerChat: prune the webhook-dedupe log daily so it can't grow unbounded.
try {
try {
const _pcClean = () => { try { db.cleanupPowerchatDeliveries(3); } catch { /* */ } };
const _pcClean = () => { try { db.cleanupPowerchatDeliveries(3); } catch { /* */ } };
@@ -249,11 +249,12 @@ function startViewerCountSweeper() {
@@ -249,11 +249,12 @@ function startViewerCountSweeper() {
}
}
// ── currency:write ───────────────────────────────────────────
// ── currency:write ───────────────────────────────────────────
+// Returns true when PowerChat accepted the event (the earn flusher re-buffers on false).
async function sendCurrencyRedemption(streamerUserId, { amount, redeemerName, rewardName, message, externalId } = {}) {
async function sendCurrencyRedemption(streamerUserId, { amount, redeemerName, rewardName, message, externalId } = {}) {
const conn = _connFor(streamerUserId, 'currency:write');
const conn = _connFor(streamerUserId, 'currency:write');
- if (!conn) return;
+ if (!conn) return false;
const amt = Math.round(Number(amount) || 0);
const amt = Math.round(Number(amount) || 0);
- if (amt < 1) return; // schema requires amount 1-1000000000
+ if (amt < 1) return false; // schema requires amount 1-1000000000
try {
try {
await oauth.apiRequest(streamerUserId, {
await oauth.apiRequest(streamerUserId, {
method: 'POST', path: '/currency-events',
method: 'POST', path: '/currency-events',
@@ -270,6 +271,7 @@ async function sendCurrencyRedemption(streamerUserId, { amount, redeemerName, re
@@ -270,6 +271,7 @@ async function sendCurrencyRedemption(streamerUserId, { amount, redeemerName, re
occurredAt: new Date().toISOString(),
occurredAt: new Date().toISOString(),
},
},
});
});
+ return true;
} catch (e) {
} catch (e) {
_noteApiError(streamerUserId, 'currency:write', e);
_noteApiError(streamerUserId, 'currency:write', e);
// "Unknown currency" means the key isn't DECLARED on the app in the PowerChat
// "Unknown currency" means the key isn't DECLARED on the app in the PowerChat
@@ -277,9 +279,62 @@ async function sendCurrencyRedemption(streamerUserId, { amount, redeemerName, re
@@ -277,9 +279,62 @@ async function sendCurrencyRedemption(streamerUserId, { amount, redeemerName, re
if (e && e.status === 400 && /unknown currency/i.test(e.message || '')) {
if (e && e.status === 400 && /unknown currency/i.test(e.message || '')) {
console.warn(`[PowerChat] currency "${CURRENCY_KEY}" is not declared on the app — declare it in the PowerChat Developer dashboard`);
console.warn(`[PowerChat] currency "${CURRENCY_KEY}" is not declared on the app — declare it in the PowerChat Developer dashboard`);
}
}
+ return false;
}
}
}
}
+// ── Channel-point EARNS → PowerChat leaderboard feed ─────────
+// Redemptions alone can't populate a leaderboard (they're rare); the board lives on
+// the steady drip of watch/chat/follow point earning. Every award is buffered per
+// (streamer, viewer) and flushed as ONE summed currency event per viewer every few
+// minutes — that keeps a busy channel inside the ~60/min currency rate limit and
+// avoids machine-gunning the overlay with micro-events.
+const EARN_FLUSH_MS = 5 * 60 * 1000;
+const EARN_MAX_PER_FLUSH = 40; // stay well under the rate limit per flush pass
+const EARN_MAX_VIEWERS = 500; // per-streamer buffer cap (drop past this, don't grow)
+const _earnBuf = new Map(); // streamerId → Map(viewerId → summed amount)
+function queueCurrencyEarn(streamerUserId, viewerUserId, amount) {
+ const amt = Math.round(Number(amount) || 0);
+ if (amt < 1 || !streamerUserId || !viewerUserId) return;
+ // Only buffer for streamers with a live currency:write connection — otherwise the
+ // buffer would grow forever for channels that never flush.
+ if (!_connFor(streamerUserId, 'currency:write')) return;
+ let m = _earnBuf.get(streamerUserId);
+ if (!m) { m = new Map(); _earnBuf.set(streamerUserId, m); }
+ if (!m.has(viewerUserId) && m.size >= EARN_MAX_VIEWERS) return;
+ m.set(viewerUserId, (m.get(viewerUserId) || 0) + amt);
+}
+async function _flushCurrencyEarns() {
+ let sent = 0;
+ for (const [streamerId, m] of _earnBuf) {
+ if (!m.size) { _earnBuf.delete(streamerId); continue; }
+ for (const [viewerId, amount] of m) {
+ if (sent >= EARN_MAX_PER_FLUSH) return; // leave the rest buffered for next pass
+ m.delete(viewerId);
+ sent++;
+ let name = null;
+ try { const u = db.getUserById(viewerId); name = u ? (u.display_name || u.username) : null; } catch { /* */ }
+ const ok = await sendCurrencyRedemption(streamerId, {
+ amount,
+ redeemerName: name || `viewer ${viewerId}`,
+ rewardName: 'Points earned',
+ externalId: `earn:${streamerId}:${viewerId}:${Date.now()}`,
+ });
+ // Transient failure → put the amount back so the points aren't lost.
+ if (!ok && _connFor(streamerId, 'currency:write')) {
+ m.set(viewerId, (m.get(viewerId) || 0) + amount);
+ }
+ }
+ }
+}
+let _earnTimer = null;
+function startCurrencyEarnFlusher() {
+ if (_earnTimer) return;
+ _earnTimer = setInterval(() => { _flushCurrencyEarns().catch(() => { }); }, EARN_FLUSH_MS);
+ if (_earnTimer.unref) _earnTimer.unref();
+ console.log('[PowerChat] channel-point earn flusher started (leaderboard feed)');
+}
+
// ── tips:write ───────────────────────────────────────────────
// ── tips:write ───────────────────────────────────────────────
// Forward an on-site Vibes donation as a MONETARY tip. amount is in Vibes UNITS;
// Forward an on-site Vibes donation as a MONETARY tip. amount is in Vibes UNITS;
// PowerChat converts server-side via the declared unitsPerUsd (100 Vibes = $1) and the
// PowerChat converts server-side via the declared unitsPerUsd (100 Vibes = $1) and the
@@ -333,5 +388,5 @@ module.exports = {
@@ -333,5 +388,5 @@ module.exports = {
CURRENCY_KEY, TIP_CURRENCY_KEY,
CURRENCY_KEY, TIP_CURRENCY_KEY,
forwardChat, forwardFollow, forwardSubscription, forwardTip,
forwardChat, forwardFollow, forwardSubscription, forwardTip,
sendViewCount, startViewerCountSweeper,
sendViewCount, startViewerCountSweeper,
- sendCurrencyRedemption, sendCustomAlert,
+ sendCurrencyRedemption, queueCurrencyEarn, startCurrencyEarnFlusher, sendCustomAlert,
};
};
@@ -31,6 +31,12 @@ const COINS = {
@@ -31,6 +31,12 @@ const COINS = {
STREAK_THRESHOLD_MIN: 60, // minutes before streak kicks in
STREAK_THRESHOLD_MIN: 60, // minutes before streak kicks in
};
};
+// Mirror every channel-point award into the streamer's PowerChat leaderboard feed
+// (batched there; a no-op unless the streamer connected PowerChat with currency:write).
+function _feedPowerchat(streamerId, userId, coins) {
+ try { require('../integrations/powerchat-platform').queueCurrencyEarn(streamerId, userId, coins); } catch { /* non-critical */ }
+}
+
// In-memory cooldown tracker (userId → lastChatCoinTime)
// In-memory cooldown tracker (userId → lastChatCoinTime)
const chatCooldowns = new Map();
const chatCooldowns = new Map();
// Bonus-game claim throttle ("userId:streamerId" → last claim ms)
// Bonus-game claim throttle ("userId:streamerId" → last claim ms)
@@ -69,6 +75,7 @@ class OpenCoins {
@@ -69,6 +75,7 @@ class OpenCoins {
}
}
const total = db.addChannelPoints(userId, streamerId, coins);
const total = db.addChannelPoints(userId, streamerId, coins);
+ _feedPowerchat(streamerId, userId, coins);
db.createCoinTransaction({
db.createCoinTransaction({
user_id: userId,
user_id: userId,
stream_id: streamId,
stream_id: streamId,
@@ -104,6 +111,7 @@ class OpenCoins {
@@ -104,6 +111,7 @@ class OpenCoins {
chatCooldowns.set(userId, now);
chatCooldowns.set(userId, now);
const total = db.addChannelPoints(userId, streamerId, COINS.CHAT_BONUS);
const total = db.addChannelPoints(userId, streamerId, COINS.CHAT_BONUS);
+ _feedPowerchat(streamerId, userId, COINS.CHAT_BONUS);
db.createCoinTransaction({
db.createCoinTransaction({
user_id: userId,
user_id: userId,
stream_id: streamId,
stream_id: streamId,
@@ -132,6 +140,7 @@ class OpenCoins {
@@ -132,6 +140,7 @@ class OpenCoins {
if (!streamerId || streamerId === userId) return null;
if (!streamerId || streamerId === userId) return null;
const total = db.addChannelPoints(userId, streamerId, COINS.FOLLOW_BONUS);
const total = db.addChannelPoints(userId, streamerId, COINS.FOLLOW_BONUS);
+ _feedPowerchat(streamerId, userId, COINS.FOLLOW_BONUS);
db.createCoinTransaction({
db.createCoinTransaction({
user_id: userId,
user_id: userId,
stream_id: null,
stream_id: null,
@@ -248,6 +257,7 @@ class OpenCoins {
@@ -248,6 +257,7 @@ class OpenCoins {
bonusClaims.set(key, now);
bonusClaims.set(key, now);
const amount = Math.max(1, (cfg.watch_amount || 10) * 3);
const amount = Math.max(1, (cfg.watch_amount || 10) * 3);
const total = db.addChannelPoints(userId, streamerId, amount);
const total = db.addChannelPoints(userId, streamerId, amount);
+ _feedPowerchat(streamerId, userId, amount);
db.createCoinTransaction({ user_id: userId, stream_id: streamId, amount, type: 'watch', message: 'Bonus game' });
db.createCoinTransaction({ user_id: userId, stream_id: streamId, amount, type: 'watch', message: 'Bonus game' });
return { coins: amount, total, streamerId };
return { coins: amount, total, streamerId };
}
}