Link to git repos

README.md

@@ -1,3 +1,11 @@
-# onion-livestreaming
+# anonstream
 
-Recipe for livestreaming over the Tor network
+Recipe for livestreaming over Tor
+
+## Repo
+
+The canonical location of this repo is https://git.076.ne.jp/ninya9k/anonstream.
+
+These mirrors exist:
+* https://gitlab.com/ninya9k/anonstream
+* https://github.com/ninya9k/anonstream

anonstream/__init__.py

@@ -89,8 +89,9 @@ def create_app(config_file):
 
     # State for tasks
     app.users_update_buffer = set()
-    app.stream_uptime = None
     app.stream_title = None
+    app.stream_uptime = None
+    app.stream_viewership = None
 
     # Background tasks' asyncio.sleep tasks, cancelled on shutdown
     app.background_sleep = set()

anonstream/helpers/user.py

@@ -76,6 +76,9 @@ def get_presence(timestamp, user):
 
     return Presence.ABSENT
 
+def is_watching(timestamp, user):
+    return get_presence(timestamp, user) == Presence.WATCHING
+
 def is_listed(timestamp, user):
     return (
         get_presence(timestamp, user)

anonstream/routes/nojs.py

@@ -2,11 +2,11 @@ from quart import current_app, request, render_template, redirect, url_for, esca
 
 from anonstream.captcha import get_random_captcha_digest_for
 from anonstream.chat import add_chat_message, Rejected
-from anonstream.stream import get_stream_title, get_stream_uptime
+from anonstream.stream import get_stream_title, get_stream_uptime, get_stream_viewership
 from anonstream.user import add_state, pop_state, try_change_appearance, verify, deverify, BadCaptcha
 from anonstream.routes.wrappers import with_user_from, render_template_with_etag
 from anonstream.helpers.chat import get_scrollback
-from anonstream.helpers.user import get_default_name
+from anonstream.helpers.user import get_default_name, Presence
 from anonstream.utils.chat import generate_nonce
 from anonstream.utils.user import concatenate_for_notice
 
@@ -19,8 +19,9 @@ async def nojs_info(user):
     return await render_template(
         'nojs_info.html',
         user=user,
-        title=await get_stream_title(),
+        viewership=get_stream_viewership(),
         uptime=get_stream_uptime(),
+        title=await get_stream_title(),
     )
 
 @current_app.route('/chat/messages.html')

anonstream/static/anonstream.js

@@ -6,7 +6,9 @@ const jsmarkup_style_color = '<style id="style-color"></style>'
 const jsmarkup_style_tripcode_display = '<style id="style-tripcode-display"></style>'
 const jsmarkup_style_tripcode_colors = '<style id="style-tripcode-colors"></style>'
 const jsmarkup_info = '<div id="info_js" data-js="true"></div>';
-const jsmarkup_info_uptime = '<aside id="info_js__uptime"></aside>';
+const jsmarkup_info_float = '<aside id="info_js__float"></aside>';
+const jsmarkup_info_float_viewership = '<div id="info_js__float__viewership"></div>';
+const jsmarkup_info_float_uptime = '<div id="info_js__float__uptime"></div>';
 const jsmarkup_info_title = '<header id="info_js__title"></header>';
 const jsmarkup_chat_messages = '<ol id="chat-messages_js" data-js="true"></ol>';
 const jsmarkup_chat_form = `\
@@ -23,7 +25,7 @@ const jsmarkup_chat_form = `\
   <input id="chat-form_js__submit" type="submit" value="Chat" accesskey="p" disabled>
 </form>`;
 
-const insert_jsmarkup = () => {
+const insert_jsmarkup = () => {jsmarkup_info_float_viewership
   if (document.getElementById("style-color") === null) {
     const parent = document.head;
     parent.insertAdjacentHTML("beforeend", jsmarkup_style_color);
@@ -40,9 +42,17 @@ const insert_jsmarkup = () => {
     const parent = document.getElementById("info");
     parent.insertAdjacentHTML("beforeend", jsmarkup_info);
   }
-  if (document.getElementById("info_js__uptime") === null) {
+  if (document.getElementById("info_js__float") === null) {
     const parent = document.getElementById("info_js");
-    parent.insertAdjacentHTML("beforeend", jsmarkup_info_uptime);
+    parent.insertAdjacentHTML("beforeend", jsmarkup_info_float);
+  }
+  if (document.getElementById("info_js__float__viewership") === null) {
+    const parent = document.getElementById("info_js__float");
+    parent.insertAdjacentHTML("beforeend", jsmarkup_info_float_viewership);
+  }
+  if (document.getElementById("info_js__float__uptime") === null) {
+    const parent = document.getElementById("info_js__float");
+    parent.insertAdjacentHTML("beforeend", jsmarkup_info_float_uptime);
   }
   if (document.getElementById("info_js__title") === null) {
     const parent = document.getElementById("info_js");
@@ -65,7 +75,8 @@ const stylesheet_tripcode_colors = document.styleSheets[3];
 
 /* create websocket */
 const info_title = document.getElementById("info_js__title");
-const info_uptime = document.getElementById("info_js__uptime");
+const info_viewership = document.getElementById("info_js__float__viewership");
+const info_uptime = document.getElementById("info_js__float__uptime");
 const chat_messages = document.getElementById("chat-messages_js");
 
 const create_chat_message = (object) => {
@@ -285,13 +296,31 @@ const chat_form_captcha_digest = document.getElementById("chat-form_js__captcha-
 const chat_form_captcha_image = document.getElementById("chat-form_js__captcha-image");
 const chat_form_captcha_answer = document.getElementById("chat-form_js__captcha-answer");
 chat_form_captcha_image.addEventListener("loadstart", (event) => {
+  chat_form_captcha_image.removeAttribute("title");
+  chat_form_captcha_image.removeAttribute("data-reloadable");
   chat_form_captcha_image.alt = "Loading...";
 });
 chat_form_captcha_image.addEventListener("load", (event) => {
   chat_form_captcha_image.removeAttribute("alt");
+  chat_form_captcha_image.dataset.reloadable = "";
+  chat_form_captcha_image.title = "Click for a new captcha";
 });
 chat_form_captcha_image.addEventListener("error", (event) => {
   chat_form_captcha_image.alt = "Captcha failed to load";
+  chat_form_captcha_image.dataset.reloadable = "";
+  chat_form_captcha_image.title = "Click for a new captcha";
+});
+chat_form_captcha_image.addEventListener("click", (event) => {
+  if (chat_form_captcha_image.dataset.reloadable === undefined) {
+    return;
+  }
+  chat_form_submit.disabled = true;
+  chat_form_captcha_image.alt = "Waiting...";
+  chat_form_captcha_image.removeAttribute("title");
+  chat_form_captcha_image.removeAttribute("data-reloadable");
+  chat_form_captcha_image.removeAttribute("src");
+  const payload = {type: "captcha"};
+  ws.send(JSON.stringify(payload));
 });
 const enable_captcha = (digest) => {
   chat_form_captcha_digest.value = digest;
@@ -302,6 +331,7 @@ const enable_captcha = (digest) => {
   chat_form_comment.required = false;
   chat_form_captcha_image.removeAttribute("src");
   chat_form_captcha_image.src = `/captcha.jpg?token=${encodeURIComponent(token)}&digest=${encodeURIComponent(digest)}`;
+  chat_form_submit.disabled = false;
   chat_form.dataset.captcha = "";
 }
 const disable_captcha = () => {
@@ -312,6 +342,7 @@ const disable_captcha = () => {
   chat_form_captcha_digest.value = "";
   chat_form_captcha_answer.value = "";
   chat_form_captcha_answer.required = false;
+  chat_form_submit.disabled = false;
   chat_form_captcha_image.removeAttribute("alt");
   chat_form_captcha_image.removeAttribute("src");
 }
@@ -352,24 +383,42 @@ const update_uptime = () => {
 }
 setInterval(update_uptime, 1000); // always update uptime
 
+const set_viewership = (n) => {
+  info_viewership.innerText = n === null ? "" : `${n} viewers`;
+}
+
 const on_websocket_message = (event) => {
   //console.log("websocket message", event);
   const receipt = JSON.parse(event.data);
   switch (receipt.type) {
     case "error":
       console.log("ws error", receipt);
+      chat_form_submit.disabled = false;
       break;
 
     case "init":
       console.log("ws init", receipt);
 
+      // set title
       set_title(receipt.title);
+
+      // set viewership
+      set_viewership(receipt.viewership);
+
+      // set uptime
       set_frozen_uptime(receipt.uptime);
       update_uptime();
 
+      // chat form nonce
       chat_form_nonce.value = receipt.nonce;
+
+      // chat form captcha digest
       receipt.digest === null ? disable_captcha() : enable_captcha(receipt.digest);
 
+      // chat form submit button
+      chat_form_submit.disabled = false;
+
+      // remove messages the server isn't acknowledging the existance of
       const seqs = new Set(receipt.messages.map((message) => {return message.seq;}));
       const to_delete = [];
       for (const chat_message of chat_messages.children) {
@@ -382,13 +431,17 @@ const on_websocket_message = (event) => {
         chat_message.remove();
       }
 
+      // settings
       default_name = receipt.default;
       max_chat_scrollback = receipt.scrollback;
+
+      // appearances
       users = receipt.users;
       update_user_names();
       update_user_colors();
       update_user_tripcodes();
 
+      // insert new messages
       const last = chat_messages.children.length == 0 ? null : chat_messages.children[chat_messages.children.length - 1];
       const last_seq = last === null ? null : parseInt(last.dataset.seq);
       for (const message of receipt.messages) {
@@ -408,6 +461,11 @@ const on_websocket_message = (event) => {
         set_frozen_uptime(receipt.uptime);
         update_uptime();
       }
+      if (receipt.viewership === 0 && frozen_uptime === null) {
+        set_viewership(null);
+      } else if (receipt.viewership !== undefined) {
+        set_viewership(receipt.viewership);
+      }
       break;
 
     case "ack":
@@ -452,6 +510,11 @@ const on_websocket_message = (event) => {
       update_user_tripcodes();
       break;
 
+    case "captcha":
+      console.log("ws captcha", receipt);
+      receipt.digest === null ? disable_captcha() : enable_captcha(receipt.digest);
+      break;
+
     default:
       console.log("incomprehensible websocket message", receipt);
   }
@@ -513,7 +576,8 @@ const chat_form_comment = document.getElementById("chat-form_js__comment");
 const chat_form_submit = document.getElementById("chat-form_js__submit");
 chat_form.addEventListener("submit", (event) => {
   event.preventDefault();
-  const payload = Object.fromEntries(new FormData(chat_form));
+  const form = Object.fromEntries(new FormData(chat_form));
+  const payload = {type: "message", form: form};
   chat_form_submit.disabled = true;
   ws.send(JSON.stringify(payload));
 });

anonstream/static/style.css

@@ -61,12 +61,18 @@ noscript {
 }
 #info_js {
   overflow-y: auto;
-  padding: 1ch 1.5ch;
+  padding: 0.75ch 1.25ch;
   height: 100%;
 }
-#info_js__uptime {
+#info_js__float {
   float: right;
   font-size: 11pt;
+  display: grid;
+  grid-auto-flow: column;
+  grid-gap: 2.5ch;
+}
+#info_js__float__uptime {
+  font-variant-numeric: tabular-nums;
 }
 #info_js__title > h1 {
   margin: 0;
@@ -181,6 +187,9 @@ noscript {
   color: inherit;
   font-size: 8pt;
 }
+#chat-form_js__captcha-image[data-reloadable] {
+  cursor: pointer;
+}
 #chat-form_js__captcha-answer {
   width: 8ch;
 }

anonstream/stream.py

@@ -4,9 +4,11 @@ import aiofiles
 from quart import current_app
 
 from anonstream.segments import get_playlist, Offline
-from anonstream.wrappers import ttl_cache_async
+from anonstream.wrappers import ttl_cache_async, with_timestamp
+from anonstream.helpers.user import is_watching
 
 CONFIG = current_app.config
+USERS = current_app.users
 
 @ttl_cache_async(CONFIG['STREAM_TITLE_CACHE_LIFETIME'])
 async def get_stream_title():
@@ -32,6 +34,14 @@ def get_stream_uptime(rounded=True):
         uptime = round(uptime, 2) if rounded else uptime
         return uptime
 
+@with_timestamp
+def get_stream_viewership(timestamp):
+    return sum(map(lambda user: is_watching(timestamp, user), USERS))
+
+def get_stream_viewership_or_none(uptime):
+    viewership = get_stream_viewership()
+    return uptime and viewership
+
 def is_online():
     try:
         get_playlist()

anonstream/tasks.py

@@ -5,7 +5,7 @@ from functools import wraps
 from quart import current_app
 
 from anonstream.broadcast import broadcast, broadcast_users_update
-from anonstream.stream import is_online, get_stream_title, get_stream_uptime
+from anonstream.stream import is_online, get_stream_title, get_stream_uptime, get_stream_viewership_or_none
 from anonstream.wrappers import with_timestamp
 from anonstream.helpers.user import is_visible
 
@@ -98,8 +98,12 @@ async def t_broadcast_users_update(iteration):
 @with_period(CONFIG['TASK_PERIOD_BROADCAST_STREAM_INFO_UPDATE'])
 async def t_broadcast_stream_info_update(iteration):
     if iteration == 0:
-        current_app.stream_title = await get_stream_title()
-        current_app.stream_uptime = get_stream_uptime()
+        title = await get_stream_title()
+        uptime = get_stream_uptime()
+        viewership = get_stream_viewership_or_none(uptime)
+        current_app.stream_title = title
+        current_app.stream_uptime = uptime
+        current_app.stream_viewership = viewership
     else:
         payload = {}
 
@@ -109,7 +113,7 @@ async def t_broadcast_stream_info_update(iteration):
             current_app.stream_title = title
             payload['title'] = title
 
-        # Check if the stream uptime has changed differently than expected
+        # Check if the stream uptime has changed more or less than expected
         if current_app.stream_uptime is None:
             expected_uptime = None
         else:
@@ -126,6 +130,12 @@ async def t_broadcast_stream_info_update(iteration):
         elif abs(uptime - expected_uptime) >= 0.0625:
             payload['uptime'] = uptime
 
+        # Check if viewership has changed
+        viewership = get_stream_viewership_or_none(uptime)
+        if current_app.stream_viewership != viewership:
+            current_app.stream_viewership = viewership
+            payload['viewership'] = viewership
+
         if payload:
             broadcast(USERS, payload={'type': 'info', **payload})
 

anonstream/templates/home.html

@@ -23,7 +23,7 @@
     <a href="#chat">chat</a>
     <a href="#both">both</a>
   </nav>
-  <footer>anonstream 1.0.0 &mdash; <a href="#" target="_blank">source</a></footer>
+  <footer>anonstream pre-1.0.0 &mdash; <a href="https://git.076.ne.jp/ninya9k/anonstream" target="_blank">source</a></footer>
   <script src="{{ url_for('static', filename='anonstream.js') }}" type="text/javascript"></script>
 </body>
 </html>

anonstream/templates/nojs_info.html

@@ -2,16 +2,23 @@
 <html>
 <head>
   <meta charset="utf-8">
+  <meta http-equiv="refresh" content="12">
   <style>
     body {
       overflow-y: auto;
-      margin: 1ch 1.5ch;
+      margin: 0.75ch 1.25ch;
       font-family: sans-serif;
       color: #ddd;
     }
-    #uptime {
+    #float {
       float: right;
       font-size: 11pt;
+      display: grid;
+      grid-auto-flow: column;
+      grid-gap: 2.5ch;
+    }
+    #float__uptime {
+      font-variant-numeric: tabular-nums;
     }
     #title > h1 {
       margin: 0;
@@ -22,7 +29,12 @@
   </style>
 </head>
 <body>
-  <aside id="uptime">{{ uptime }}</aside>
+  {% if uptime is not none %}
+    <aside id="float">
+      <div id="float__viewership">{{ viewership }} viewers</div>
+      <div id="float__uptime">{{ uptime }}</div>
+    </aside>
+  {% endif %}
   <header id="title"><h1>{{ title }}</h1></header>
 </body>
 </html>

anonstream/utils/websocket.py

@@ -1,19 +1,31 @@
 class Malformed(Exception):
     pass
 
+def get(t, pairs, key, default=None):
+    value = pairs.get(key, default)
+    if isinstance(value, t):
+        return value
+    else:
+        raise Malformed(f'malformed {key}')
+
 def parse_websocket_data(receipt):
     if not isinstance(receipt, dict):
         raise Malformed('not a json object')
 
-    comment = receipt.get('comment')
-    if not isinstance(comment, str):
-        raise Malformed('malformed comment')
+    match receipt.get('type'):
+        case 'message':
+            form = get(dict, receipt, 'form')
+            nonce = get(str, form, 'nonce')
+            comment = get(str, form, 'comment')
+            digest = get(str, form, 'captcha-digest', '')
+            answer = get(str, form, 'captcha-answer', '')
+            return nonce, comment, digest, answer
 
-    nonce = receipt.get('nonce')
-    if not isinstance(nonce, str):
-        raise Malformed('malformed nonce')
+        case 'appearance':
+            raise NotImplemented
 
-    digest = receipt.get('captcha-digest', '')
-    answer = receipt.get('captcha-answer', '')
+        case 'captcha':
+            return None
 
-    return nonce, comment, digest, answer
+        case _:
+            raise Malformed('malformed type')

anonstream/websocket.py

@@ -3,7 +3,7 @@ import json
 
 from quart import current_app, websocket
 
-from anonstream.stream import get_stream_title, get_stream_uptime
+from anonstream.stream import get_stream_title, get_stream_uptime, get_stream_viewership_or_none
 from anonstream.captcha import get_random_captcha_digest_for
 from anonstream.chat import get_all_messages_for_websocket, add_chat_message, Rejected
 from anonstream.user import get_all_users_for_websocket, see, verify, deverify, BadCaptcha
@@ -13,11 +13,14 @@ from anonstream.utils.websocket import parse_websocket_data, Malformed
 CONFIG = current_app.config
 
 async def websocket_outbound(queue, user):
+    uptime = get_stream_uptime()
+    viewership = get_stream_viewership_or_none(uptime)
     payload = {
         'type': 'init',
         'nonce': generate_nonce(),
         'title': await get_stream_title(),
-        'uptime': get_stream_uptime(),
+        'uptime': uptime,
+        'viewership': viewership,
         'messages': get_all_messages_for_websocket(),
         'users': get_all_users_for_websocket(),
         'default': {
@@ -41,7 +44,7 @@ async def websocket_inbound(queue, user):
         finally:
             see(user)
         try:
-            nonce, comment, digest, answer = parse_websocket_data(receipt)
+            parsed = parse_websocket_data(receipt)
         except Malformed as e:
             error , *_ = e.args
             payload = {
@@ -49,29 +52,47 @@ async def websocket_inbound(queue, user):
                 'because': error,
             }
         else:
-            try:
-                verification_happened = verify(user, digest, answer)
-            except BadCaptcha as e:
-                notice, *_ = e.args
-            else:
-                try:
-                    message_was_added = add_chat_message(
-                        user,
-                        nonce,
-                        comment,
-                        ignore_empty=verification_happened,
-                    )
-                except Rejected as e:
-                    notice, *_ = e.args
-                else:
-                    deverify(user)
-                    notice = None
-            payload = {
-                'type': 'ack',
-                'nonce': nonce,
-                'next': generate_nonce(),
-                'notice': notice,
-                'clear': message_was_added,
-                'digest': get_random_captcha_digest_for(user),
-            }
+            match parsed:
+                case [nonce, comment, digest, answer]:
+                    payload = handle_inbound_message(user, *parsed)
+
+                case None:
+                    payload = handle_inbound_captcha(user)
+
         queue.put_nowait(payload)
+
+def handle_inbound_captcha(user):
+    return {
+        'type': 'captcha',
+        'digest': get_random_captcha_digest_for(user),
+    }
+
+def handle_inbound_message(user, nonce, comment, digest, answer):
+    try:
+        verification_happened = verify(user, digest, answer)
+    except BadCaptcha as e:
+        notice, *_ = e.args
+        message_was_added = False
+    else:
+        try:
+            message_was_added = add_chat_message(
+                user,
+                nonce,
+                comment,
+                ignore_empty=verification_happened,
+            )
+        except Rejected as e:
+            notice, *_ = e.args
+            message_was_added = False
+        else:
+            notice = None
+            if message_was_added:
+                deverify(user)
+    return {
+        'type': 'ack',
+        'nonce': nonce,
+        'next': generate_nonce(),
+        'notice': notice,
+        'clear': message_was_added,
+        'digest': get_random_captcha_digest_for(user),
+    }