# End-to-end proof of PostActivityChannel (examples/blog/app/channels/
# post_activity_channel.rb) - the WebSocket + Hub broadcaster from lib/
# prism/websocket wired into a real controller for the first time.
# Everything below happens over a real TCPSocket doing its own HTTP
# Upgrade handshake by hand (same style as test_websocket_integration.rb),
# alongside ordinary Net::HTTP requests hitting the SAME running server -
# proving the two protocols coexist on one Prism::Server, and that an
# actual CommentsController#approve/#reject/#destroy over HTTP is what
# pushes the broadcast, not a hand-triggered Hub call.
$LOAD_PATH.unshift(File.expand_path('lib', __dir__))
require 'prism'
require 'net/http'
require 'socket'
require 'fileutils'
require 'json'

ROOT = __dir__
SCRATCH_DB_DIR = File.join(ROOT, 'examples/blog/db/websocket_demo')
FileUtils.rm_rf(SCRATCH_DB_DIR)
ENV['PRISM_BLOG_DB_DIR'] = SCRATCH_DB_DIR

require_relative 'examples/blog/app'

# max_websocket_connections: 3 mirrors examples/blog/bin/server (which uses
# a realistic 2000) but small enough that this demo's own three concurrent
# viewers below (viewer_a, viewer_b, bystander) can actually saturate it,
# proving the cap for real instead of just asserting the option is accepted.
server = Prism::Server.new(Blog, port: 0, max_websocket_connections: 3).start
port = server.port
puts "blog_websocket_demo server on port #{port}"
puts '-' * 40

Frame = Prism::WebSocket::Frame
Handshake = Prism::WebSocket::Handshake

def cookie_from(response)
  lines = response.get_fields('set-cookie') || []
  lines.map { |line| line.split(';').first }.join('; ')
end

def csrf_token_from(html)
  html[/name="csrf-token" content="([^"]+)"/, 1]
end

# Hand-rolled Upgrade handshake over a raw TCPSocket, deliberately not
# reusing anything from Net::HTTP - an Upgrade request is not an ordinary
# HTTP request/response cycle (see Prism::Server#serve intercepting it
# before @app.call ever runs), so this proves the real wire protocol,
# the same way test_websocket_integration.rb does for the library itself.
def open_cable(host, port, post_id)
  socket = TCPSocket.new(host, port)
  key = [Array.new(16) { rand(256) }.pack('C16')].pack('m0')
  socket.write(
    "GET /cable?post_id=#{post_id} HTTP/1.1\r\n" \
    "Host: #{host}\r\n" \
    "Upgrade: websocket\r\n" \
    "Connection: Upgrade\r\n" \
    "Sec-WebSocket-Key: #{key}\r\n" \
    "Sec-WebSocket-Version: 13\r\n" \
    "\r\n"
  )
  headers = +''
  until headers.end_with?("\r\n\r\n")
    line = socket.gets("\r\n")
    raise 'connection closed mid-handshake' if line.nil?
    headers << line
  end
  raise "expected a 101 handshake, got: #{headers}" unless headers.include?('101')
  expected_accept = Handshake.accept_key(key)
  raise 'Sec-WebSocket-Accept did not match the RFC 6455 formula' unless headers.include?(expected_accept)
  socket
end

# Same handshake request as open_cable, but for proving max_websocket_
# connections rejects it BEFORE any Upgrade: the server must answer with an
# ordinary HTTP 503 (not a 101, and not a partial/dropped connection), read
# off the raw socket exactly like open_cable reads a successful handshake.
def attempt_rejected_cable(host, port, post_id)
  socket = TCPSocket.new(host, port)
  key = [Array.new(16) { rand(256) }.pack('C16')].pack('m0')
  socket.write(
    "GET /cable?post_id=#{post_id} HTTP/1.1\r\n" \
    "Host: #{host}\r\n" \
    "Upgrade: websocket\r\n" \
    "Connection: Upgrade\r\n" \
    "Sec-WebSocket-Key: #{key}\r\n" \
    "Sec-WebSocket-Version: 13\r\n" \
    "\r\n"
  )
  headers = +''
  until headers.end_with?("\r\n\r\n")
    line = socket.gets("\r\n")
    raise 'connection closed before any HTTP status line arrived' if line.nil?
    headers << line
  end
  socket.close
  headers
end

def read_event(socket)
  JSON.parse(Frame.decode(socket).payload)
end

Net::HTTP.start('127.0.0.1', port) do |http|
  home = http.get('/')
  cookie = cookie_from(home)
  token = csrf_token_from(home.body)
  puts "GET / -> #{home.code}"

  signup_body = URI.encode_www_form(
    'user[email]' => 'moderator@example.com', 'user[password]' => 'correct horse battery staple',
    'user[password_confirmation]' => 'correct horse battery staple', 'authenticity_token' => token
  )
  signup = http.post('/users', signup_body, 'Content-Type' => 'application/x-www-form-urlencoded', 'Cookie' => cookie)
  raise "expected sign-up to redirect, got #{signup.code}" unless signup.code == '302'
  cookie = cookie_from(signup)
  puts "POST /users (sign up as moderator) -> #{signup.code}"

  create_body = URI.encode_www_form(
    'post[title]' => 'Live comment counts', 'post[body]' => 'Watching this page update without a reload.',
    'authenticity_token' => token
  )
  create = http.post('/posts', create_body, 'Content-Type' => 'application/x-www-form-urlencoded', 'Cookie' => cookie)
  raise "expected the post to be created, got #{create.code}" unless create.code == '302'
  post_path = create['location']
  post_id = post_path.split('/').last
  cookie = cookie_from(create)
  puts "POST /posts -> #{create.code} Location=#{post_path}"

  # Two independent viewers open the post's live channel before any
  # comment exists - each gets its own subscribed acknowledgement, proving
  # the Hub keys subscribers by stream (this post's id), not just "anyone
  # connected right now".
  viewer_a = open_cable('127.0.0.1', port, post_id)
  viewer_b = open_cable('127.0.0.1', port, post_id)
  welcome_a = read_event(viewer_a)
  welcome_b = read_event(viewer_b)
  raise "expected viewer_a's subscribe ack, got #{welcome_a.inspect}" unless welcome_a == { 'event' => 'subscribed', 'post_id' => post_id }
  raise "expected viewer_b's subscribe ack, got #{welcome_b.inspect}" unless welcome_b == { 'event' => 'subscribed', 'post_id' => post_id }
  puts "GET /cable?post_id=#{post_id} x2 -> 101 Switching Protocols, both subscribed"

  # A third connection opens a channel for a DIFFERENT (nonexistent) post -
  # it must never see this post's events, proving streams are genuinely
  # isolated per post, not one shared firehose.
  bystander = open_cable('127.0.0.1', port, '999999')
  read_event(bystander) # subscribe ack, discarded
  puts "GET /cable?post_id=999999 -> 101 (unrelated post, must stay silent below)"

  # viewer_a, viewer_b, and bystander are now three live WebSocket
  # connections on this process - exactly the max_websocket_connections
  # cap set above. A fourth, entirely unrelated stream must be turned away
  # BEFORE the Upgrade handshake even starts, proving the limit is global
  # (per-process) rather than PostActivityChannel::MAX_VIEWERS_PER_POST's
  # per-stream cap, which would never trip with only three viewers total.
  rejection_headers = attempt_rejected_cable('127.0.0.1', port, post_id)
  raise "expected a 503, got: #{rejection_headers}" unless rejection_headers.include?('503')
  puts "GET /cable?post_id=#{post_id} (4th concurrent WS connection) -> 503 Service Unavailable " \
       '(max_websocket_connections: 3 reached, rejected before any Upgrade)'

  comment_body = URI.encode_www_form('comment[author]' => 'Reader', 'comment[body]' => 'First!', 'authenticity_token' => token)
  comment = http.post("/posts/#{post_id}/comments", comment_body,
                       'Content-Type' => 'application/x-www-form-urlencoded', 'Cookie' => cookie)
  raise "expected the comment to be created, got #{comment.code}" unless comment.code == '302'
  cookie = cookie_from(comment)
  show = http.get(post_path, 'Cookie' => cookie)
  comment_id = show.body[/id="comment-(\d+)"/, 1]
  raise 'could not find the new comment id on the post page' unless comment_id
  puts "POST /posts/#{post_id}/comments -> #{comment.code} (still pending, no broadcast yet)"

  approve_body = URI.encode_www_form('_method' => 'PATCH', 'authenticity_token' => token)
  approve = http.post("/posts/#{post_id}/comments/#{comment_id}/approve", approve_body,
                       'Content-Type' => 'application/x-www-form-urlencoded', 'Cookie' => cookie)
  raise "expected the approval to redirect, got #{approve.code}" unless approve.code == '302'
  cookie = cookie_from(approve)

  event_a = read_event(viewer_a)
  event_b = read_event(viewer_b)
  expected = { 'event' => 'comment_count', 'post_id' => post_id, 'count' => 1 }
  raise "viewer_a expected #{expected.inspect}, got #{event_a.inspect}" unless event_a == expected
  raise "viewer_b expected #{expected.inspect}, got #{event_b.inspect}" unless event_b == expected
  puts "POST /posts/#{post_id}/comments/#{comment_id}/approve -> #{approve.code}, " \
       "both live viewers pushed {event: comment_count, count: 1} over the real socket"

  reject_body = URI.encode_www_form('_method' => 'PATCH', 'authenticity_token' => token)
  reject = http.post("/posts/#{post_id}/comments/#{comment_id}/reject", reject_body,
                      'Content-Type' => 'application/x-www-form-urlencoded', 'Cookie' => cookie)
  raise "expected the rejection to redirect, got #{reject.code}" unless reject.code == '302'
  cookie = cookie_from(reject)

  event_a = read_event(viewer_a)
  expected = { 'event' => 'comment_count', 'post_id' => post_id, 'count' => 0 }
  raise "viewer_a expected #{expected.inspect} after rejection, got #{event_a.inspect}" unless event_a == expected
  puts "POST /posts/#{post_id}/comments/#{comment_id}/reject -> #{reject.code}, " \
       "live viewer pushed {event: comment_count, count: 0} (approved -> rejected demotes the count back down)"

  # Re-approve, then destroy outright - the third mutation that can move
  # the visible count, and the last one this demo has left to prove.
  http.post("/posts/#{post_id}/comments/#{comment_id}/approve", approve_body,
            'Content-Type' => 'application/x-www-form-urlencoded', 'Cookie' => cookie)
  read_event(viewer_a) # count -> 1 again, already proven above

  destroy_body = URI.encode_www_form('_method' => 'DELETE', 'authenticity_token' => token)
  destroy = http.post("/posts/#{post_id}/comments/#{comment_id}", destroy_body,
                       'Content-Type' => 'application/x-www-form-urlencoded', 'Cookie' => cookie)
  raise "expected the destroy to redirect, got #{destroy.code}" unless destroy.code == '302'

  event_a = read_event(viewer_a)
  event_b = read_event(viewer_b)
  expected = { 'event' => 'comment_count', 'post_id' => post_id, 'count' => 0 }
  raise "viewer_a expected #{expected.inspect} after destroy, got #{event_a.inspect}" unless event_a == expected
  raise "viewer_b expected #{expected.inspect} after destroy, got #{event_b.inspect}" unless event_b == expected
  puts "DELETE /posts/#{post_id}/comments/#{comment_id} -> #{destroy.code}, " \
       'both live viewers pushed {event: comment_count, count: 0} for the removed comment'

  bystander.write(Frame.encode(Frame::PING, '', mask: true))
  reply = Frame.decode(bystander)
  raise "expected the unrelated channel's socket to only ever see our own PONG, got opcode #{reply.opcode}" unless reply.opcode == Frame::PONG
  puts "bystander subscribed to post_id=999999 received ZERO comment_count events for post #{post_id} " \
       '(streams are isolated per post, confirmed by the only frame available being our own PONG)'

  viewer_a.write(Frame.encode(Frame::CLOSE, '', mask: true))
  close_reply = Frame.decode(viewer_a)
  raise "expected a CLOSE frame echoed back, got opcode #{close_reply.opcode}" unless close_reply.opcode == Frame::CLOSE
  puts 'viewer_a sent a close frame -> server echoed CLOSE back (RFC 6455 5.5.1), on_close unsubscribed it from the Hub'

  # examples/blog/app.rb registers '/cable' with max_message_size: 2048,
  # far below the library's 16 MiB default - proven here the same way
  # test_websocket_integration.rb proves the option in the library itself,
  # but this time through the actual application's real configuration
  # instead of a purpose-built test endpoint.
  oversized_viewer = open_cable('127.0.0.1', port, post_id)
  read_event(oversized_viewer) # subscribe ack, discarded
  oversized_viewer.write(Frame.encode(Frame::TEXT, 'x' * 3000, mask: true))
  close_frame = Frame.decode(oversized_viewer)
  raise "expected a CLOSE frame, got opcode #{close_frame.opcode}" unless close_frame.opcode == Frame::CLOSE
  close_code = close_frame.payload.byteslice(0, 2).unpack1('n')
  raise "expected close code 1009 (message too big), got #{close_code}" unless close_code == 1009
  puts "sent a 3000 byte frame on /cable (max_message_size: 2048 in examples/blog/app.rb) " \
       "-> server closed with code #{close_code} (message too big), without disturbing viewer_b or bystander"
  oversized_viewer.close

  [viewer_a, viewer_b, bystander].each { |socket| socket.close unless socket.closed? }

  still_alive = http.get('/')
  raise "expected the plain HTTP server to still be healthy after the WebSocket traffic, got #{still_alive.code}" unless still_alive.code == '200'
  puts "GET / -> #{still_alive.code} (ordinary HTTP routes on the same server are unaffected by the cable traffic above)"
end

server.stop
puts '-' * 40
puts 'blog_websocket_demo done: PostActivityChannel subscribed two real WebSocket clients to one post ' \
     'and kept a third, unrelated one silent; three genuine HTTP moderation actions (approve/reject/destroy) ' \
     'each pushed a live {event: comment_count} frame to every subscriber over the real socket, and the close ' \
     'handshake correctly unsubscribed a departing viewer from the shared Hub.'