<commit_before>{EventEmitter} = require('events')

async = require('async')  # don't delete even if not used below, since this needs to be available to page/
_     = require('underscore')

salvus_version = require('salvus_version')

diffsync = require('diffsync')

message = require("message")
misc    = require("misc")

docs = require("docs")

defaults = misc.defaults
required = defaults.required

# This is the default time in *seconds* between pings sent by the
# client to the server to indicate that a given session is being
# actively viewed.  This variable is used by hub.coffee to set
# its kill timeout, so do not change the name here without changing
# it there.   It *is* safe to change the value.
exports.DEFAULT_SESSION_PING_TIME = 600

# JSON_CHANNEL is the channel used for JSON.  The hub imports this
# file, so if this constant is ever changed (for some reason?), it
# only has to be changed on this one line.  Moreover, channel
# assignment in the hub is implemented *without* the assumption that
# the JSON channel is '\u0000'.
JSON_CHANNEL = '\u0000'
exports.JSON_CHANNEL = JSON_CHANNEL # export, so can be used by hub


# change these soon
git0 = 'git0'
gitls = 'git-ls'

class Session extends EventEmitter
    # events:
    #    - 'open'   -- session is initialized, open and ready to be used
    #    - 'close'  -- session's connection is closed/terminated
    #    - 'execute_javascript' -- code that server wants client to run related to this session
    constructor: (opts) ->
        opts = defaults opts,
            conn         : required     # a Connection instance
            project_id   : required
            session_uuid : required
            params       : undefined
            data_channel : undefined    # optional extra channel that is used for raw data

        @start_time   = misc.walltime()
        @conn         = opts.conn
        @params       = opts.params
        @project_id   = opts.project_id
        @session_uuid = opts.session_uuid
        @data_channel = opts.data_channel
        @emit("open")

        if @reconnect?
            @conn.on "connected", (() => setTimeout(@reconnect, 500))

    reconnect: (cb) =>
        # Called when the connection gets dropped, then reconnects
        if not @conn._signed_in? or not @conn._signed_in
            setTimeout(@reconnect, 500)
            return  # do *NOT* do cb?() yet!

        @conn.call
            message : message.connect_to_session
                session_uuid : @session_uuid
                type         : @type()
                project_id   : @project_id
                params       : @params
            cb      : (error, reply) =>
                if error
                    cb?(error); return
                switch reply.event
                    when 'error'
                        cb?(reply.error)
                    when 'session_connected'
                        @conn.change_data_channel
                            prev_channel : @data_channel
                            new_channel  : reply.data_channel
                            session      : @
                        @data_channel = reply.data_channel
                        cb?()
                    else
                        cb?("bug in hub")

    terminate_session: (cb) =>
        @conn.call
            message :
                message.terminate_session
                    project_id   : @project_id
                    session_uuid : @session_uuid
            timeout : 30
            cb      : cb

    walltime: () =>
        return misc.walltime() - @start_time

    handle_data: (data) =>
        @emit("data", data)

    write_data: (data) ->
        @conn.write_data(@data_channel, data)

    # default = SIGINT
    interrupt: (cb) ->
        tm = misc.mswalltime()
        if @_last_interrupt? and tm - @_last_interrupt < 100
            # client self-limit: do not send signals too frequently, since that wastes bandwidth and can kill the process
            cb?()
        else
            @_last_interrupt = tm
            @conn.call(message:message.send_signal(session_uuid:@session_uuid, signal:2), timeout:10, cb:cb)

    kill: (cb) ->
        @emit("close")
        @conn.call(message:message.send_signal(session_uuid:@session_uuid, signal:9), timeout:10, cb:cb)

    restart: (cb) =>
        @conn.call(message:message.restart_session(session_uuid:@session_uuid), timeout:10, cb:cb)

    # Starts a ping interval timer that periodicially pings the server
    # to indicate that this session is being actively viewed.  Pinging
    # stops if the function continue_pinging() returns false.
    # If the continue_pinging function is not defined, just ping server once.
    ping: (continue_pinging) ->
        if not continue_pinging?
            @conn.send(message.ping_session(session_uuid:@session_uuid))
            return
        timer = undefined
        ping = () =>
            if continue_pinging()
                @ping()
            else
                clearInterval(timer)
        timer = setInterval(ping, exports.DEFAULT_SESSION_PING_TIME * 1000)

###
#
# A Sage session, which links the client to a running Sage process;
# provides extra functionality to kill/interrupt, etc.
#
#   Client <-- (sockjs) ---> Hub  <--- (tcp) ---> sage_server
#
###

class SageSession extends Session
    # If cb is given, it is called every time output for this particular code appears;
    # No matter what, you can always still listen in with the 'output' even, and note
    # the uuid, which is returned from this function.
    execute_code: (opts) ->
        opts = defaults opts,
            code     : required
            cb       : undefined
            data     : undefined
            preparse : true
            uuid     : undefined

        if opts.uuid?
            uuid = opts.uuid
        else
            uuid = misc.uuid()
        if opts.cb?
            @conn.execute_callbacks[uuid] = opts.cb

        @conn.send(
            message.execute_code
                id   : uuid
                code : opts.code
                data : opts.data
                session_uuid : @session_uuid
                preparse : opts.preparse
        )

        return uuid

    type: () => "sage"

    introspect: (opts) ->
        opts.session_uuid = @session_uuid
        @conn.introspect(opts)

# TODO -- for 'interact2'
#
#     variable: (opts) ->
#         opts = defaults opts,
#             name      : required
#             namespace : 'globals()'
#         return SageSessionVariable(@, opts.name, opts.namespace)

# class SageSessionVariable extends EventEmitter
#     constructor: (@session, @name, @namespace, cb) ->
#         @uuid = misc.uuid()
#         @session.execute_code
#             code : "sage_salvus.register_variable(salvus.data['name'], eval(salvus.data['namespace']), salvus.data['uuid'])"
#             data :
#                 name      : @name
#                 namespace : @namespace
#                 uuid      : @uuid
#             preparse : false
#             cb       : cb

#     set : (value, cb) =>
#         @session.execute_code
#             code     :
#             cb       :
#             preparse : false

#     get : (cb) =>  # cb(err, value) -- value must be JSON-able
#         @session.execute_code
#             code :
#             cb   :
#             preparse : false


###
#
# A Console session, which connects the client to a pty on a remote machine.
#
#   Client <-- (sockjs) ---> Hub  <--- (tcp) ---> console_server
#
###

class ConsoleSession extends Session
    type: () => "console"




class exports.Connection extends EventEmitter
    # Connection events:
    #    - 'connecting' -- trying to establish a connection
    #    - 'connected'  -- succesfully established a connection; data is the protocol as a string
    #    - 'error'      -- called when an error occurs
    #    - 'output'     -- received some output for stateless execution (not in any session)
    #    - 'execute_javascript' -- code that server wants client to run (not for a particular session)
    #    - 'ping'       -- a pong is received back; data is the round trip ping time
    #    - 'message'    -- emitted when a JSON message is received           on('message', (obj) -> ...)
    #    - 'data'       -- emitted when raw data (not JSON) is received --   on('data, (id, data) -> )...
    #    - 'signed_in'  -- server pushes a succesful sign in to the client (e.g., due to
    #                      'remember me' functionality); data is the signed_in message.
    #    - 'project_list_updated' -- sent whenever the list of projects owned by this user
    #                      changed; data is empty -- browser could ignore this unless
    #                      the project list is currently being displayed.
    #    - 'project_data_changed - sent when data about a specific project has changed,
    #                      e.g., title/description/settings/etc.



    constructor: (@url) ->
        @setMaxListeners(100)   #TODO: lower this to <=10 and track down issues/remove leaks.
        @emit("connecting")
        @_id_counter       = 0
        @_sessions         = {}
        @_new_sessions     = {}
        @_data_handlers    = {}
        @execute_callbacks = {}
        @call_callbacks    = {}

        @register_data_handler(JSON_CHANNEL, @handle_json_data)

        # IMPORTANT! Connection is an abstract base class.  Derived classes must
        # implement a method called _connect that takes a URL and a callback, and connects to
        # the SockJS server with that url, then creates the following event emitters:
        #      "connected", "error", "close"
        # and returns a function to write raw data to the socket.

        @_connect @url, (data) =>
            if data.length > 0  # all messages must start with a channel; length 0 means nothing.

                # Incoming messages are tagged with a single UTF-16
                # character c (there are 65536 possibilities).  If
                # that character is JSON_CHANNEL, the message is
                # encoded as JSON and we handle it in the usual way.
                # If the character is anything else, the raw data in
                # the message is sent to an appropriate handler, if
                # one has previously been registered.  The motivation
                # is that we the ability to multiplex multiple
                # sessions over a *single* SockJS connection, and it
                # is absolutely critical that there is minimal
                # overhead regarding the amount of data transfered --
                # 1 character is minimal!

                channel = data[0]
                data    = data.slice(1)

                @_handle_data(channel, data)

                # give other listeners a chance to do something with this data.
                @emit("data", channel, data)

        @_last_pong = misc.walltime()
        @_connected = false

        # have to get a round trip packet between client and hub every
        # this many ms, or client will start freaking out and trying
        # to reconnect.  This limits things like max size of files we
        # can edit.
        @_ping_check_interval = 15000  # 15 seconds.
        @_ping_check_id = setInterval((()=>@ping(); @_ping_check()), @_ping_check_interval)

    close: () ->
        clearInterval(@_ping_check_id)
        @_conn.close()

    _ping_check: () ->
        if @_connected and (@_last_ping - @_last_pong > 1.1*@_ping_check_interval/1000.0)
            @_signed_in = false
            @_fix_connection?()

    # Send a JSON message to the hub server.
    send: (mesg) ->
        @write_data(JSON_CHANNEL, misc.to_json(mesg))

    # Send raw data via certain channel to the hub server.
    write_data: (channel, data) =>
        try
            @_write(channel + data)
        catch err
            # TODO: this happens when trying to send and the client not connected
            # We might save up messages in a local queue and keep retrying, for
            # a sort of offline mode ?  I have not worked out how to handle this yet.
            #console.log(err)

    handle_json_data: (data) =>
        mesg = misc.from_json(data)
        switch mesg.event
            when "execute_javascript"
                if mesg.session_uuid?
                    @_sessions[mesg.session_uuid].emit("execute_javascript", mesg)
                else
                    @emit("execute_javascript", mesg)
            when "output"
                cb = @execute_callbacks[mesg.id]
                if cb?
                    cb(mesg)
                    delete @execute_callbacks[mesg.id] if mesg.done
                if mesg.session_uuid?  # executing in a persistent session
                    @_sessions[mesg.session_uuid].emit("output", mesg)
                else   # stateless exec
                    @emit("output", mesg)
            when "terminate_session"
                session = @_sessions[mesg.session_uuid]
                session?.emit("close")
            when "session_reconnect"
                if mesg.data_channel?
                    @_sessions[mesg.data_channel]?.reconnect()
                else if mesg.session_uuid?
                    @_sessions[mesg.session_uuid]?.reconnect()
            when "pong"
                @_last_pong = misc.walltime()
                @emit("ping", @_last_pong - @_last_ping)
            when "cookies"
                @_cookies?(mesg)
            when "signed_in"
                @account_id = mesg.account_id
                @_signed_in = true
                @emit("signed_in", mesg)
            when "project_list_updated", 'project_data_changed'
                @emit(mesg.event, mesg)
            when "codemirror_diffsync_ready"
                @emit(mesg.event, mesg)
            when "codemirror_bcast"
                @emit(mesg.event, mesg)
            when "error"
                # An error that isn't tagged with an id -- some sort of general problem.
                if not mesg.id?
                    alert_message(type:"error", message:mesg.error)
                    return

        id = mesg.id  # the call f(null,mesg) can mutate mesg (!), so we better save the id here.
        f = @call_callbacks[id]<commit_msg>errors -- change something from visible alert to console log<commit_after>{EventEmitter} = require('events')

async = require('async')  # don't delete even if not used below, since this needs to be available to page/
_     = require('underscore')

salvus_version = require('salvus_version')

diffsync = require('diffsync')

message = require("message")
misc    = require("misc")

docs = require("docs")

defaults = misc.defaults
required = defaults.required

# This is the default time in *seconds* between pings sent by the
# client to the server to indicate that a given session is being
# actively viewed.  This variable is used by hub.coffee to set
# its kill timeout, so do not change the name here without changing
# it there.   It *is* safe to change the value.
exports.DEFAULT_SESSION_PING_TIME = 600

# JSON_CHANNEL is the channel used for JSON.  The hub imports this
# file, so if this constant is ever changed (for some reason?), it
# only has to be changed on this one line.  Moreover, channel
# assignment in the hub is implemented *without* the assumption that
# the JSON channel is '\u0000'.
JSON_CHANNEL = '\u0000'
exports.JSON_CHANNEL = JSON_CHANNEL # export, so can be used by hub


# change these soon
git0 = 'git0'
gitls = 'git-ls'

class Session extends EventEmitter
    # events:
    #    - 'open'   -- session is initialized, open and ready to be used
    #    - 'close'  -- session's connection is closed/terminated
    #    - 'execute_javascript' -- code that server wants client to run related to this session
    constructor: (opts) ->
        opts = defaults opts,
            conn         : required     # a Connection instance
            project_id   : required
            session_uuid : required
            params       : undefined
            data_channel : undefined    # optional extra channel that is used for raw data

        @start_time   = misc.walltime()
        @conn         = opts.conn
        @params       = opts.params
        @project_id   = opts.project_id
        @session_uuid = opts.session_uuid
        @data_channel = opts.data_channel
        @emit("open")

        if @reconnect?
            @conn.on "connected", (() => setTimeout(@reconnect, 500))

    reconnect: (cb) =>
        # Called when the connection gets dropped, then reconnects
        if not @conn._signed_in? or not @conn._signed_in
            setTimeout(@reconnect, 500)
            return  # do *NOT* do cb?() yet!

        @conn.call
            message : message.connect_to_session
                session_uuid : @session_uuid
                type         : @type()
                project_id   : @project_id
                params       : @params
            cb      : (error, reply) =>
                if error
                    cb?(error); return
                switch reply.event
                    when 'error'
                        cb?(reply.error)
                    when 'session_connected'
                        @conn.change_data_channel
                            prev_channel : @data_channel
                            new_channel  : reply.data_channel
                            session      : @
                        @data_channel = reply.data_channel
                        cb?()
                    else
                        cb?("bug in hub")

    terminate_session: (cb) =>
        @conn.call
            message :
                message.terminate_session
                    project_id   : @project_id
                    session_uuid : @session_uuid
            timeout : 30
            cb      : cb

    walltime: () =>
        return misc.walltime() - @start_time

    handle_data: (data) =>
        @emit("data", data)

    write_data: (data) ->
        @conn.write_data(@data_channel, data)

    # default = SIGINT
    interrupt: (cb) ->
        tm = misc.mswalltime()
        if @_last_interrupt? and tm - @_last_interrupt < 100
            # client self-limit: do not send signals too frequently, since that wastes bandwidth and can kill the process
            cb?()
        else
            @_last_interrupt = tm
            @conn.call(message:message.send_signal(session_uuid:@session_uuid, signal:2), timeout:10, cb:cb)

    kill: (cb) ->
        @emit("close")
        @conn.call(message:message.send_signal(session_uuid:@session_uuid, signal:9), timeout:10, cb:cb)

    restart: (cb) =>
        @conn.call(message:message.restart_session(session_uuid:@session_uuid), timeout:10, cb:cb)

    # Starts a ping interval timer that periodicially pings the server
    # to indicate that this session is being actively viewed.  Pinging
    # stops if the function continue_pinging() returns false.
    # If the continue_pinging function is not defined, just ping server once.
    ping: (continue_pinging) ->
        if not continue_pinging?
            @conn.send(message.ping_session(session_uuid:@session_uuid))
            return
        timer = undefined
        ping = () =>
            if continue_pinging()
                @ping()
            else
                clearInterval(timer)
        timer = setInterval(ping, exports.DEFAULT_SESSION_PING_TIME * 1000)

###
#
# A Sage session, which links the client to a running Sage process;
# provides extra functionality to kill/interrupt, etc.
#
#   Client <-- (sockjs) ---> Hub  <--- (tcp) ---> sage_server
#
###

class SageSession extends Session
    # If cb is given, it is called every time output for this particular code appears;
    # No matter what, you can always still listen in with the 'output' even, and note
    # the uuid, which is returned from this function.
    execute_code: (opts) ->
        opts = defaults opts,
            code     : required
            cb       : undefined
            data     : undefined
            preparse : true
            uuid     : undefined

        if opts.uuid?
            uuid = opts.uuid
        else
            uuid = misc.uuid()
        if opts.cb?
            @conn.execute_callbacks[uuid] = opts.cb

        @conn.send(
            message.execute_code
                id   : uuid
                code : opts.code
                data : opts.data
                session_uuid : @session_uuid
                preparse : opts.preparse
        )

        return uuid

    type: () => "sage"

    introspect: (opts) ->
        opts.session_uuid = @session_uuid
        @conn.introspect(opts)

# TODO -- for 'interact2'
#
#     variable: (opts) ->
#         opts = defaults opts,
#             name      : required
#             namespace : 'globals()'
#         return SageSessionVariable(@, opts.name, opts.namespace)

# class SageSessionVariable extends EventEmitter
#     constructor: (@session, @name, @namespace, cb) ->
#         @uuid = misc.uuid()
#         @session.execute_code
#             code : "sage_salvus.register_variable(salvus.data['name'], eval(salvus.data['namespace']), salvus.data['uuid'])"
#             data :
#                 name      : @name
#                 namespace : @namespace
#                 uuid      : @uuid
#             preparse : false
#             cb       : cb

#     set : (value, cb) =>
#         @session.execute_code
#             code     :
#             cb       :
#             preparse : false

#     get : (cb) =>  # cb(err, value) -- value must be JSON-able
#         @session.execute_code
#             code :
#             cb   :
#             preparse : false


###
#
# A Console session, which connects the client to a pty on a remote machine.
#
#   Client <-- (sockjs) ---> Hub  <--- (tcp) ---> console_server
#
###

class ConsoleSession extends Session
    type: () => "console"




class exports.Connection extends EventEmitter
    # Connection events:
    #    - 'connecting' -- trying to establish a connection
    #    - 'connected'  -- succesfully established a connection; data is the protocol as a string
    #    - 'error'      -- called when an error occurs
    #    - 'output'     -- received some output for stateless execution (not in any session)
    #    - 'execute_javascript' -- code that server wants client to run (not for a particular session)
    #    - 'ping'       -- a pong is received back; data is the round trip ping time
    #    - 'message'    -- emitted when a JSON message is received           on('message', (obj) -> ...)
    #    - 'data'       -- emitted when raw data (not JSON) is received --   on('data, (id, data) -> )...
    #    - 'signed_in'  -- server pushes a succesful sign in to the client (e.g., due to
    #                      'remember me' functionality); data is the signed_in message.
    #    - 'project_list_updated' -- sent whenever the list of projects owned by this user
    #                      changed; data is empty -- browser could ignore this unless
    #                      the project list is currently being displayed.
    #    - 'project_data_changed - sent when data about a specific project has changed,
    #                      e.g., title/description/settings/etc.



    constructor: (@url) ->
        @setMaxListeners(100)   #TODO: lower this to <=10 and track down issues/remove leaks.
        @emit("connecting")
        @_id_counter       = 0
        @_sessions         = {}
        @_new_sessions     = {}
        @_data_handlers    = {}
        @execute_callbacks = {}
        @call_callbacks    = {}

        @register_data_handler(JSON_CHANNEL, @handle_json_data)

        # IMPORTANT! Connection is an abstract base class.  Derived classes must
        # implement a method called _connect that takes a URL and a callback, and connects to
        # the SockJS server with that url, then creates the following event emitters:
        #      "connected", "error", "close"
        # and returns a function to write raw data to the socket.

        @_connect @url, (data) =>
            if data.length > 0  # all messages must start with a channel; length 0 means nothing.

                # Incoming messages are tagged with a single UTF-16
                # character c (there are 65536 possibilities).  If
                # that character is JSON_CHANNEL, the message is
                # encoded as JSON and we handle it in the usual way.
                # If the character is anything else, the raw data in
                # the message is sent to an appropriate handler, if
                # one has previously been registered.  The motivation
                # is that we the ability to multiplex multiple
                # sessions over a *single* SockJS connection, and it
                # is absolutely critical that there is minimal
                # overhead regarding the amount of data transfered --
                # 1 character is minimal!

                channel = data[0]
                data    = data.slice(1)

                @_handle_data(channel, data)

                # give other listeners a chance to do something with this data.
                @emit("data", channel, data)

        @_last_pong = misc.walltime()
        @_connected = false

        # have to get a round trip packet between client and hub every
        # this many ms, or client will start freaking out and trying
        # to reconnect.  This limits things like max size of files we
        # can edit.
        @_ping_check_interval = 15000  # 15 seconds.
        @_ping_check_id = setInterval((()=>@ping(); @_ping_check()), @_ping_check_interval)

    close: () ->
        clearInterval(@_ping_check_id)
        @_conn.close()

    _ping_check: () ->
        if @_connected and (@_last_ping - @_last_pong > 1.1*@_ping_check_interval/1000.0)
            @_signed_in = false
            @_fix_connection?()

    # Send a JSON message to the hub server.
    send: (mesg) ->
        @write_data(JSON_CHANNEL, misc.to_json(mesg))

    # Send raw data via certain channel to the hub server.
    write_data: (channel, data) =>
        try
            @_write(channel + data)
        catch err
            # TODO: this happens when trying to send and the client not connected
            # We might save up messages in a local queue and keep retrying, for
            # a sort of offline mode ?  I have not worked out how to handle this yet.
            #console.log(err)

    handle_json_data: (data) =>
        mesg = misc.from_json(data)
        switch mesg.event
            when "execute_javascript"
                if mesg.session_uuid?
                    @_sessions[mesg.session_uuid].emit("execute_javascript", mesg)
                else
                    @emit("execute_javascript", mesg)
            when "output"
                cb = @execute_callbacks[mesg.id]
                if cb?
                    cb(mesg)
                    delete @execute_callbacks[mesg.id] if mesg.done
                if mesg.session_uuid?  # executing in a persistent session
                    @_sessions[mesg.session_uuid].emit("output", mesg)
                else   # stateless exec
                    @emit("output", mesg)
            when "terminate_session"
                session = @_sessions[mesg.session_uuid]
                session?.emit("close")
            when "session_reconnect"
                if mesg.data_channel?
                    @_sessions[mesg.data_channel]?.reconnect()
                else if mesg.session_uuid?
                    @_sessions[mesg.session_uuid]?.reconnect()
            when "pong"
                @_last_pong = misc.walltime()
                @emit("ping", @_last_pong - @_last_ping)
            when "cookies"
                @_cookies?(mesg)
            when "signed_in"
                @account_id = mesg.account_id
                @_signed_in = true
                @emit("signed_in", mesg)
            when "project_list_updated", 'project_data_changed'
                @emit(mesg.event, mesg)
            when "codemirror_diffsync_ready"
                @emit(mesg.event, mesg)
            when "codemirror_bcast"
                @emit(mesg.event, mesg)
            when "error"
                # An error that isn't tagged with an id -- some sort of general problem.
                if not mesg.id?
                    console.log("WARNING: #{mesg.error}")
                    return

        id = mesg.id  # the call f(null,mesg) can mutate mesg (!), so we better save the id here.
        f = @call_callbacks[id]
{'fraction_non_alphanumeric': 0.08883435582822086, 'fraction_numerical': 0.004837861524978089, 'mean_word_length': 1.7548044422984066, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 77, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2089801', 'n_tokens_mistral': 7890, 'n_tokens_neox': 7500, 'n_words': 2483}