Module: Tina4::Realtime

Defined in:
lib/tina4/realtime.rb,
lib/tina4/realtime/channel.rb,
lib/tina4/realtime/message.rb,
lib/tina4/realtime/storage.rb,
lib/tina4/realtime/workspace.rb,
lib/tina4/realtime/attachment.rb,
lib/tina4/realtime/s3_storage.rb,
lib/tina4/realtime/local_storage.rb,
lib/tina4/realtime/channel_member.rb,
lib/tina4/realtime/storage_backend.rb

Overview

Real-time collaboration mount for Tina4 (Ruby), parity with the Python master's tina4_python.realtime. A zero-dependency control plane for building Slack/Teams-class tools:

  • calls: a WebRTC signalling relay + self-describing ICE-config endpoint. Media is peer-to-peer (mesh); Tina4 carries no media, it only relays the offer/answer/ICE handshake. An SFU drops in later with no route changes.
  • chat: persistent channels + messages (framework-owned ORM models), a secured chat WebSocket with live presence / typing / read receipts, and a history endpoint for catch-up-on-reconnect.
  • files: uploads/downloads through a pluggable StorageBackend.

Paths are convention defaults, overridable via prefix, and the client discovers the resolved paths from the config endpoint so client and server never drift.

Tina4::Realtime.mount                                  # calls only
Tina4::Realtime.mount(features: %w[calls chat])        # add persistent chat
Tina4::Realtime.mount(prefix: "/api/collab", features: %w[calls chat files])

Env: TINA4_RTC_BACKEND (default mesh), TINA4_RTC_STUN_URLS, TINA4_RTC_TURN_URL + TINA4_RTC_TURN_SECRET (ephemeral coturn creds), TINA4_RTC_TURN_TTL, plus the TINA4_STORAGE_* set for files.

Defined Under Namespace

Modules: Storage Classes: Attachment, Channel, ChannelMember, LocalStorage, Message, S3Storage, StorageBackend, Workspace

Constant Summary collapse

DEFAULT_STUN = 
"stun:stun.l.google.com:19302"

Class Method Summary collapse

Class Method Details

.authorized?(identity, channel_id) ⇒ Boolean

Shared membership guard for chat channels and file access.

Returns:

  • (Boolean) 


280
281
282
283
284
285
286
287
288
 
# File 'lib/tina4/realtime.rb', line 280

def authorized?(identity, channel_id)
  return false if identity.nil?
  return !!@authorize.call(identity, channel_id) unless @authorize.nil?

  ChannelMember.count("channel_id = ? AND user_id = ?", [channel_id, identity]).positive?
rescue StandardError => e
  Tina4::Log.error("realtime chat authorize check failed: #{e.message}")
  false
end

.chat_handler(connection, event, data) ⇒ Object

Secured chat WebSocket. Ruby fires (connection, event, data); event is a Symbol (:open / :message / :close).



210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
 
# File 'lib/tina4/realtime.rb', line 210

def chat_handler(connection, event, data)
  raw = connection.params[:channel].to_s
  return unless raw.match?(/\A\d+\z/) # framework channels are addressed by integer id

  channel_id = raw.to_i
  identity = identity(connection.auth)
  key = "chat:#{channel_id}"

  case event
  when :open
    unless authorized?(identity, channel_id)
      connection.send_json({ "type" => "error", "error" => "not a member of this channel" })
      connection.close
      return
    end
    connection.join_room(key)
    connection.send_json({ "type" => "presence", "event" => "roster", "users" => roster(key) })
    connection.broadcast_to_room(
      key, { "type" => "presence", "event" => "join", "user_id" => identity }.to_json, exclude_self: true
    )
  when :close
    connection.broadcast_to_room(
      key, { "type" => "presence", "event" => "leave", "user_id" => identity }.to_json, exclude_self: true
    )
  when :message
    # Re-check membership on every inbound frame: it can be revoked
    # mid-session, and we never trust a payload's identity.
    return unless authorized?(identity, channel_id)

    payload = parse_json(data)
    return unless payload.is_a?(Hash)

    kind = payload["type"] || "message"
    case kind
    when "typing"
      connection.broadcast_to_room(
        key, { "type" => "typing", "user_id" => identity }.to_json, exclude_self: true
      )
    when "read"
      mark_read(channel_id, identity)
      connection.broadcast_to_room(
        key, { "type" => "read", "user_id" => identity, "at" => now_iso }.to_json, exclude_self: true
      )
    when "message"
      body = (payload["body"] || "").to_s.strip
      return if body.empty?

      saved = persist_message(channel_id, identity, body, payload["thread_id"])
      # Deliver to everyone INCLUDING the sender so the sender's optimistic
      # message is reconciled with its server id + timestamp.
      connection.broadcast_to_room(key, { "type" => "message", "message" => saved }.to_json) if saved
    end
  end
end

.ensure_chat_tables ⇒ Object 



385
386
387
388
389
390
391
392
393
394
 
# File 'lib/tina4/realtime.rb', line 385

def ensure_chat_tables
  [Workspace, Channel, ChannelMember, Message, Attachment].each(&:create_table)
  true
rescue StandardError => e
  Tina4::Log.error(
    "realtime chat could not create its tables (#{e.message}). Chat needs a bound " \
    "database - call Tina4.bind_database(db) or set TINA4_DATABASE_URL before mount."
  )
  false
end

.form_value(request, name) ⇒ Object 



300
301
302
303
304
305
 
# File 'lib/tina4/realtime.rb', line 300

def form_value(request, name)
  body = request.body
  return body[name] if body.is_a?(Hash) && body.key?(name)

  (request.query && request.query[name]) || (request.params && request.params[name.to_sym])
end

.history(channel_id, query) ⇒ Object

Messages for a channel, newest-first, paged by a before cursor. The standard infinite-scroll-backwards shape a chat client pages with.



348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
 
# File 'lib/tina4/realtime.rb', line 348

def history(channel_id, query)
  limit = (query["limit"] || query[:limit] || 50).to_i
  limit = 50 if limit <= 0
  limit = [[limit, 1].max, 200].min
  before = query["before"] || query[:before]

  clause = "channel_id = ?"
  params = [channel_id]
  unless before.nil? || before.to_s.empty?
    clause += " AND id < ?"
    params << before.to_i
  end

  rows = Message.where(clause, params)
  rows = rows.sort_by { |m| -(m.id || 0) }.first(limit)
  rows.map do |m|
    {
      "id" => m.id, "channel_id" => m.channel_id, "user_id" => m.user_id,
      "body" => m.body, "thread_id" => m.thread_id, "created_at" => m.created_at
    }
  end
end

.ice_servers ⇒ Object

Build the ICE server list from the environment. Always includes STUN. Adds a TURN entry with time-limited credentials (coturn use-auth-secret scheme: username = , credential = base64(HMAC-SHA1(secret, username))) when both TINA4_RTC_TURN_URL and TINA4_RTC_TURN_SECRET are set.



51
52
53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/tina4/realtime.rb', line 51

def ice_servers
  stun = ENV["TINA4_RTC_STUN_URLS"] || DEFAULT_STUN
  servers = [{ "urls" => split_urls(stun) }]

  turn_url = ENV["TINA4_RTC_TURN_URL"]
  secret = ENV["TINA4_RTC_TURN_SECRET"]
  if turn_url && !turn_url.empty? && secret && !secret.empty?
    ttl = (ENV["TINA4_RTC_TURN_TTL"] || "3600").to_i
    username = (Time.now.to_i + ttl).to_s
    credential = Base64.strict_encode64(OpenSSL::HMAC.digest("SHA1", secret, username))
    servers << { "urls" => split_urls(turn_url), "username" => username, "credential" => credential }
  end
  servers
end

.identity(auth) ⇒ Object

Extract a stable user identity from a verified JWT payload. Chat expects one of user_id / sub / id. Returns a string, or nil when unauthenticated.



269
270
271
272
273
274
275
276
277
 
# File 'lib/tina4/realtime.rb', line 269

def identity(auth)
  return nil unless auth.is_a?(Hash)

  %w[user_id sub id].each do |claim|
    val = auth[claim] || auth[claim.to_sym]
    return val.to_s unless val.nil?
  end
  nil
end

.mark_read(channel_id, identity) ⇒ Object 



336
337
338
339
340
341
342
343
344
 
# File 'lib/tina4/realtime.rb', line 336

def mark_read(channel_id, identity)
  member = ChannelMember.where("channel_id = ? AND user_id = ?", [channel_id, identity]).first
  return if member.nil?

  member.last_read_at = now_iso
  member.save
rescue StandardError => e
  Tina4::Log.error("realtime chat: read-receipt update failed: #{e.message}")
end

.mount(prefix: "", authorize: nil, storage: nil, features: nil) ⇒ Object

Mount the realtime surface and return the resolved path map (also served from the config endpoint so the client can discover it).

Parameters:

  • prefix (String) (defaults to: "") —

    mount the whole surface under this path

  • authorize (Proc) (defaults to: nil) —

    membership guard authorize.(identity, channel_id) for chat/files; defaults to a ChannelMember check

  • storage (StorageBackend) (defaults to: nil) —

    for the files feature

  • features (Array<String>) (defaults to: nil) —

    any of "calls", "chat", "files"



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/tina4/realtime.rb', line 74

def mount(prefix: "", authorize: nil, storage: nil, features: nil)
  features = features || ["calls"]
  p = prefix.to_s.gsub(%r{\A/+|/+\z}, "")
  p = p.empty? ? "" : "/#{p}"
  @authorize = authorize
  @storage = storage

  # Chat and files both persist through the framework-owned chat models.
  ensure_chat_tables if features.include?("chat") || features.include?("files")

  paths = { "backend" => "mesh" }
  if features.include?("calls")
    paths["config"] = "#{p}/api/rtc/config"
    paths["signalling"] = "#{p}/ws/rtc"
  end
  if features.include?("chat")
    paths["config"] ||= "#{p}/api/rtc/config"
    paths["chat"] = "#{p}/ws/chat"
    paths["messages"] = "#{p}/api/channels"
  end
  if features.include?("files")
    paths["config"] ||= "#{p}/api/rtc/config"
    paths["files"] = "#{p}/api/files"
  end

  register_config(paths, features) if paths.key?("config")
  register_calls(paths) if features.include?("calls")
  register_chat(paths) if features.include?("chat")
  register_files(paths, storage) if features.include?("files")

  paths
end

.now_iso ⇒ Object 



315
316
317
 
# File 'lib/tina4/realtime.rb', line 315

def now_iso
  Time.now.utc.strftime("%Y-%m-%dT%H:%M:%SZ")
end

.parse_json(data) ⇒ Object 



307
308
309
310
311
312
313
 
# File 'lib/tina4/realtime.rb', line 307

def parse_json(data)
  return data if data.is_a?(Hash)

  JSON.parse(data.to_s)
rescue JSON::ParserError, TypeError
  nil
end

.persist_message(channel_id, identity, body, thread_id) ⇒ Object 



319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
 
# File 'lib/tina4/realtime.rb', line 319

def persist_message(channel_id, identity, body, thread_id)
  thread = thread_id.nil? || thread_id == "" ? nil : thread_id.to_i
  created_at = now_iso
  msg = Message.new(
    channel_id: channel_id, user_id: identity.to_s, body: body,
    thread_id: thread, created_at: created_at
  )
  if msg.save == false
    Tina4::Log.error("realtime chat: failed to persist message in channel #{channel_id}")
    return nil
  end
  {
    "id" => msg.id, "channel_id" => channel_id, "user_id" => identity.to_s,
    "body" => body, "thread_id" => thread, "created_at" => created_at
  }
end

.register_calls(paths) ⇒ Object 



125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/tina4/realtime.rb', line 125

def register_calls(paths)
  # WebRTC signalling relay (mesh). Tina4 never parses the SDP; peers filter
  # by `to`. The signalling room is public - the media is E2E between peers.
  Tina4::Router.websocket("#{paths['signalling']}/{room}") do |connection, event, data|
    room = connection.params[:room].to_s
    next if room.empty?

    key = "rtc:#{room}"
    case event
    when :open
      connection.join_room(key)
    when :message
      connection.broadcast_to_room(key, data, exclude_self: true)
    end
  end
end

.register_chat(paths) ⇒ Object 



142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/tina4/realtime.rb', line 142

def register_chat(paths)
  Tina4::Router.websocket("#{paths['chat']}/{channel}") { |c, e, d| chat_handler(c, e, d) }.secure

  Tina4::Router.get("#{paths['messages']}/{id}/messages") do |request, response|
    identity = Tina4::Realtime.identity(request.user)
    channel_id = request.params[:id].to_i
    if channel_id <= 0
      response.json({ "error" => "invalid channel id" }, Tina4::HTTP_BAD_REQUEST)
    elsif !Tina4::Realtime.authorized?(identity, channel_id)
      response.json({ "error" => "forbidden" }, Tina4::HTTP_FORBIDDEN)
    else
      response.json(Tina4::Realtime.history(channel_id, request.query || {}), Tina4::HTTP_OK)
    end
  end.secure
end

.register_config(paths, features) ⇒ Object

----- route registration -------------------------------------------------



109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/tina4/realtime.rb', line 109

def register_config(paths, features)
  Tina4::Router.get(paths["config"]) do |_request, response|
    body = { "backend" => "mesh" }
    if features.include?("calls")
      body["iceServers"] = Tina4::Realtime.ice_servers
      body["signalling"] = "#{paths['signalling']}/{room}"
    end
    if features.include?("chat")
      body["chat"] = "#{paths['chat']}/{channel}"
      body["messages"] = "#{paths['messages']}/{id}/messages"
    end
    body["files"] = paths["files"] if features.include?("files")
    response.json(body, Tina4::HTTP_OK)
  end
end

.register_files(paths, storage) ⇒ Object 



158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
 
# File 'lib/tina4/realtime.rb', line 158

def register_files(paths, storage)
  store = Tina4::Realtime::Storage.select(storage)

  # POST is auth-required by default (mirrors GET being public).
  Tina4::Router.post(paths["files"]) do |request, response|
    identity = Tina4::Realtime.identity(request.user)
    channel_id = Tina4::Realtime.form_value(request, "channel_id").to_i
    if channel_id <= 0
      response.json({ "error" => "channel_id is required" }, Tina4::HTTP_BAD_REQUEST)
    elsif !Tina4::Realtime.authorized?(identity, channel_id)
      response.json({ "error" => "forbidden" }, Tina4::HTTP_FORBIDDEN)
    else
      upload = request.files && request.files["file"]
      if upload.nil?
        response.json({ "error" => "no file uploaded (field 'file')" }, Tina4::HTTP_BAD_REQUEST)
      else
        saved = Tina4::Realtime.store_upload(store, channel_id, upload)
        saved["url"] = store.url(saved["key"]) || "#{paths['files']}/#{saved['key']}"
        response.json(saved, Tina4::HTTP_CREATED)
      end
    end
  end

  Tina4::Router.get("#{paths['files']}/{key}") do |request, response|
    identity = Tina4::Realtime.identity(request.user)
    key = request.params[:key].to_s
    att = Attachment.where("storage_key = ?", [key]).first
    if att.nil?
      response.json({ "error" => "not found" }, Tina4::HTTP_NOT_FOUND)
    elsif !Tina4::Realtime.authorized?(identity, att.channel_id.to_i)
      response.json({ "error" => "forbidden" }, Tina4::HTTP_FORBIDDEN)
    else
      direct = store.url(key)
      if direct
        response.redirect(direct, 302)
      else
        data = store.get(key)
        if data.nil?
          response.json({ "error" => "not found" }, Tina4::HTTP_NOT_FOUND)
        else
          response.header("Content-Disposition", "inline; filename=\"#{att.filename || key}\"")
          response.call(data, Tina4::HTTP_OK, att.mime || "application/octet-stream")
        end
      end
    end
  end.secure
end

.roster(key) ⇒ Object

Presence roster: the distinct authenticated identities currently in a room, as strings (sorted). Collected from the live connections on the manager.



292
293
294
295
296
297
298
 
# File 'lib/tina4/realtime.rb', line 292

def roster(key)
  mgr = Tina4::WebSocket.current
  return [] if mgr.nil?

  seen = mgr.get_room_connections(key).map { |c| identity(c.auth) }.compact
  seen.uniq.sort
end

.split_urls(csv) ⇒ Object 



396
397
398
 
# File 'lib/tina4/realtime.rb', line 396

def split_urls(csv)
  csv.split(",").map(&:strip).reject(&:empty?)
end

.store_upload(store, channel_id, upload) ⇒ Object 



371
372
373
374
375
376
377
378
379
380
381
382
383
 
# File 'lib/tina4/realtime.rb', line 371

def store_upload(store, channel_id, upload)
  filename = upload["filename"] || upload[:filename] || "file"
  mime = upload["type"] || upload[:type] || "application/octet-stream"
  content = upload["content"] || upload[:content] || ""
  key = Tina4::Realtime::Storage.key(filename)
  store.put(key, content, mime)
  att = Attachment.new(
    channel_id: channel_id, storage_key: key, filename: filename,
    mime: mime, size: content.bytesize
  )
  att.save
  { "id" => att.id, "key" => key, "filename" => filename, "mime" => mime, "size" => content.bytesize }
end