1/*  Part of SWISH
    2
    3    Author:        Jan Wielemaker
    4    E-mail:        jan@swi-prolog.org
    5    WWW:           http://www.swi-prolog.org
    6    Copyright (C): 2016-2024, VU University Amsterdam
    7                              CWI Amsterdam
    8                              SWI-Prolog Solutions b.v.
    9    All rights reserved.
   10
   11    Redistribution and use in source and binary forms, with or without
   12    modification, are permitted provided that the following conditions
   13    are met:
   14
   15    1. Redistributions of source code must retain the above copyright
   16       notice, this list of conditions and the following disclaimer.
   17
   18    2. Redistributions in binary form must reproduce the above copyright
   19       notice, this list of conditions and the following disclaimer in
   20       the documentation and/or other materials provided with the
   21       distribution.
   22
   23    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
   24    "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
   25    LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
   26    FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
   27    COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
   28    INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
   29    BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
   30    LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
   31    CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
   32    LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
   33    ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
   34    POSSIBILITY OF SUCH DAMAGE.
   35*/
   36
   37:- module(swish_chat,
   38          [ chat_broadcast/1,           % +Message
   39            chat_broadcast/2,           % +Message, +Channel
   40            chat_to_profile/2,          % +ProfileID, :HTML
   41            chat_about/2,               % +DocID, +Message
   42
   43            notifications//1,           % +Options
   44            broadcast_bell//1           % +Options
   45          ]).   46:- use_module(library(http/hub)).   47:- use_module(library(http/http_dispatch)).   48:- use_module(library(http/http_session)).   49:- use_module(library(http/http_parameters)).   50:- use_module(library(http/http_cors)).   51:- use_module(library(http/websocket)).   52:- use_module(library(http/json)).   53:- use_module(library(error)).   54:- use_module(library(lists)).   55:- use_module(library(option)).   56:- use_module(library(debug)).   57:- use_module(library(uuid)).   58:- use_module(library(random)).   59:- use_module(library(base64)).   60:- use_module(library(apply)).   61:- use_module(library(broadcast)).   62:- use_module(library(ordsets)).   63:- use_module(library(http/html_write)).   64:- use_module(library(http/http_path)).   65:- if(exists_source(library(user_profile))).   66:- use_module(library(user_profile)).   67:- endif.   68:- use_module(library(aggregate)).   69:- use_module(library(redis)).   70:- use_module(library(solution_sequences)).   71
   72:- use_module(storage).   73:- use_module(gitty).   74:- use_module(config).   75:- use_module(avatar).   76:- use_module(noble_avatar).   77:- use_module(chatstore).   78:- use_module(authenticate).   79:- use_module(pep).   80:- use_module(content_filter).   81:- use_module(swish_redis).   82
   83:- html_meta(chat_to_profile(+, html)).

   99:- multifile swish_config:config/2.  100
  101swish_config:config(hangout, 'Hangout.swinb').
  102swish_config:config(avatars, svg).              % or 'noble'
  103swish_config:config(session_lost_timeout, 300).
  104
  105
  106                 /*******************************
  107                 *      ESTABLISH WEBSOCKET     *
  108                 *******************************/
  109
  110:- http_handler(swish(chat), start_chat, [ id(swish_chat) ]).  111
  112:- meta_predicate must_succeed(0).

 start_chat(+Request)

HTTP handler that establishes a websocket connection where a user gets an avatar and optionally a name.

  119start_chat(Request) :-
  120    memberchk(method(options), Request),
  121    !,
  122    cors_enable(Request,
  123                [ methods([get])
  124                ]),
  125    format('~n').
  126start_chat(Request) :-
  127    cors_enable,
  128    authenticate(Request, Identity),
  129    start_chat(Request, [identity(Identity)]).
  130
  131start_chat(Request, Options) :-
  132    authorized(chat(open), Options),
  133    (   http_in_session(Session)
  134    ->  CheckLogin = false
  135    ;   http_open_session(Session, []),
  136        CheckLogin = true
  137    ),
  138    check_flooding(Session),
  139    http_parameters(Request,
  140                    [ avatar(Avatar, [optional(true)]),
  141                      nickname(NickName, [optional(true)]),
  142                      reconnect(Token, [optional(true)])
  143                    ]),
  144    extend_options([ avatar(Avatar),
  145                     nick_name(NickName),
  146                     reconnect(Token),
  147                     check_login(CheckLogin)
  148                   ], Options, ChatOptions),
  149    debug(chat(websocket), 'Accepting (session ~p)', [Session]),
  150    http_upgrade_to_websocket(
  151        accept_chat(Session, ChatOptions),
  152        [ guarded(false),
  153          subprotocols(['v1.chat.swish.swi-prolog.org', chat])
  154        ],
  155        Request).
  156
  157extend_options([], Options, Options).
  158extend_options([H|T0], Options, [H|T]) :-
  159    ground(H),
  160    !,
  161    extend_options(T0, Options, T).
  162extend_options([_|T0], Options, T) :-
  163    extend_options(T0, Options, T).

 check_flooding(+Session)

See whether the client associated with a session is flooding us and if so, return a resource error.

  171check_flooding(Session) :-
  172    get_time(Now),
  173    (   http_session_retract(websocket(Score, Last))
  174    ->  Passed is Now-Last,
  175        NewScore is Score*(2**(-Passed/60)) + 10
  176    ;   NewScore = 10,
  177        Passed = 0
  178    ),
  179    debug(chat(flooding), 'Flooding score: ~2f (session ~p)',
  180          [NewScore, Session]),
  181    http_session_assert(websocket(NewScore, Now)),
  182    (   NewScore > 50
  183    ->  throw(http_reply(resource_error(
  184                             error(permission_error(reconnect, websocket,
  185                                                    Session),
  186                                   websocket(reconnect(Passed, NewScore))))))
  187    ;   true
  188    ).

 accept_chat(+Session, +Options, +WebSocket)

Create the websocket for the chat session. If the websocket was lost due to a network failure, the client will try to reconnect with a reconnect token. If this is successful we restore the old identity and WSID. If not, we add the websocket to the "hub" and send a a welcome message over the established websocket.

  198accept_chat(Session, Options, WebSocket) :-
  199    must_succeed(accept_chat_(Session, Options, WebSocket)),
  200    Long is 100*24*3600,                        % see update_session_timeout/1
  201    http_set_session(Session, timeout(Long)).
  202
  203accept_chat_(Session, Options, WebSocket) :-
  204    create_chat_room,
  205    (   option(reconnect(Token), Options),
  206        http_session_data(wsid(WSID, Token), Session),
  207        wsid_status_del_lost(WSID),
  208        existing_visitor(WSID, Session, TmpUser, UserData),
  209        must_succeed(hub_add(swish_chat, WebSocket, WSID))
  210    ->  Reason = rejoined
  211    ;   must_succeed(hub_add(swish_chat, WebSocket, WSID)),
  212        random_key(16, Token),
  213        http_session_asserta(wsid(WSID, Token), Session),
  214        must_succeed(create_visitor(WSID, Session, TmpUser, UserData, Options)),
  215        Reason = joined
  216    ),
  217    gc_visitors,
  218    visitor_count(Visitors),
  219    option(check_login(CheckLogin), Options, true),
  220    Msg0 = _{ type:welcome,
  221	      uid:TmpUser,
  222	      wsid:WSID,
  223	      reconnect:Token,
  224	      visitors:Visitors,
  225	      check_login:CheckLogin
  226	    },
  227    add_redis_consumer(Msg0, Msg),
  228    AckMsg = UserData.put(Msg),
  229    (   hub_send(WSID, json(AckMsg))
  230    ->  must_succeed(chat_broadcast(UserData.put(_{type:Reason,
  231                                                   visitors:Visitors,
  232                                                   wsid:WSID}))),
  233        debug(chat(websocket), '~w (session ~p, wsid ~p)',
  234              [Reason, Session, WSID])
  235    ;   Reason = joined
  236    ->  debug(chat(websocket), 'Failed to acknowledge join for ~p in ~p',
  237              [WSID, Session]),
  238        http_session_retractall(wsid(WSID, Token), Session),
  239        reclaim_visitor(WSID),
  240        fail
  241    ;   debug(chat(websocket), 'Failed to acknowledge rejoin for ~p in ~p',
  242              [WSID, Session]),
  243        fail
  244    ).
  245
  246add_redis_consumer(Msg0, Msg) :-
  247    use_redis,
  248    redis_consumer(Consumer),
  249    !,
  250    Msg = Msg0.put(consumer, Consumer).
  251add_redis_consumer(Msg, Msg).
  252
  253must_succeed(Goal) :-
  254    catch_with_backtrace(Goal, E, (print_message(warning, E), fail)),
  255    !.
  256must_succeed(Goal) :-
  257    print_message(warning, goal_failed(Goal)),
  258    fail.
  259
  260
  261                 /*******************************
  262                 *              DATA            *
  263                 *******************************/
  264
  265/* - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
  266We have three user identifications:
  267
  268  - The WSID (Web Socket ID).  This uniquely identifies an open SWISH
  269    window.
  270  - The HTTP session id.   A single browser may have multiple SWISH
  271    windows open and thus be associated with multiple WSIDs.
  272  - A Visitor ID.  This captures elementary knowledge of the user
  273    associated with the session.  Each session has exactly on Visitor
  274    id.
  275
  276Redis DB organization
  277
  278  - swish:chat:wsid
  279    Redis set of known WSID
  280  - swish:chat:session:WSID -> at(Consumer,Session,Token)
  281    Expresses that WSID belongs to the SWISH server identified by
  282    Consumer, the given HTTP session and if if it is lost it can
  283    be reastablished using Token.
  284  - swish:chat:lost:WSID -> Time
  285    We lost connection to WSID at Time (e.g., websocket disconnect)
  286  - swish:chat:unload:WSID -> boolean
  287    If `true`, the page was gracefully unloaded.
  288  - swish:chat:visitor:Visitor -> UserData
  289    Visitor is a UUID reflecting a visitor with properties for
  290    identification.
  291
  292In addition, we store data on the session:
  293
  294  - websocket(Score, Time)
  295    Keeps a score based on attempts to establish the websocket.  Used
  296    to deny a connection request with a 503 error
  297  - swish_user(Visitor)
  298    Connect the session to the given visitor UUID.
  299- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - */

 wsid_status(?WSID, ?Status)

 wsid_session(?WSID, ?SessionId)

 session_user(?Session, ?TmpUser)

 visitor_data(?TmpUser, ?UserData:dict)

 subscription(?WSID, ?Channel, ?SubChannel)

These predicates represent our notion of visitors. Active modes:

• wsid_status(+WSID, -Status)
  • Indexes: 1
  • retract(wsid_status(+WSID, -Status))
  • assertz(wsid_status(+WSID, +Status))
  • retractall(wsid_status(+WSID, _))
• wsid_session(?WSID, ?SessionId)
  • Indexes: 1,2,3
• session_user(?Session, ?TmpUser)
  • Indexes: 1
• visitor_data(?TmpUser, ?UserData)
  • Indexes: 1
• subscription(?WSID, ?Channel, ?SubChannel)

  Indexes:1,3

  Currently all subscriptions are on the Channel gitty

Arguments:

WSID	- is the identifier of the web socket. As we may have to reconnect lost connections, this is may be replaced.
Session	- is the session identifier. This is used to connect SWISH actions to WSIDs.
TmpUser	- is the ID with which we identify the user for this run. The value is a UUID and thus doesn't reveal the real identity of the user.
UserData	- is a dict that holds information about the real user identity. This can be empty if no information is known about this user.
Status	- is one of unload or lost(Time)
Channel	- is an atom denoting a chat channel
SubChannel	- is a related sub channel.

  339:- dynamic
  340    wsid_status_db/2,            % WSID, Status
  341    wsid_session_db/2,           % WSID, Session
  342    session_user_db/2,		    % Session, TmpUser
  343    visitor_data_db/2,		    % TmpUser, Data
  344    subscription_db/3.		    % WSID, Channel, SubChannel

 redis_key(+Which, -Server, -Key) is semidet

 redis_key_ro(+Which, -Server, -Key) is semidet

Find the Redis server and key for a query. The redis_key_ro/3 variant returns the nearby Redis replica if it exists. This can only be used with read-only keys.

  354redis_key(Which, Server, Key) :-
  355    swish_config(redis, Server),
  356    swish_config(redis_prefix, Prefix),
  357    Which =.. List,
  358    atomic_list_concat([Prefix, chat | List], :, Key).
  359
  360redis_key_ro(Which, Server, Key) :-
  361    swish_config(redis_ro, Server),
  362    !,
  363    swish_config(redis_prefix, Prefix),
  364    Which =.. List,
  365    atomic_list_concat([Prefix, chat | List], :, Key).
  366redis_key_ro(Which, Server, Key) :-
  367    redis_key(Which, Server, Key).
  368
  369use_redis :-
  370    swish_config(redis, _).

 wsid_status(+WSID, -Status)

Status is one of lost(Time) if we lost contact at Time or unload if the websocket was cleanly disconnected.

The Redis version keeps two keys per WSID as described below. Note that these keys only exist on temporary lost or disconnecting websockets.

• unload:WSID = boolean
• lost:WSID = time

  385wsid_status(WSID, Status) :-
  386    redis_key_ro(unload(WSID), Server, UnloadKey),
  387    !,
  388    redis_key_ro(lost(WSID), Server, LostKey),
  389    redis(Server,
  390          [ get(UnloadKey) -> Unload,
  391            get(LostKey) -> Lost
  392          ]),
  393    (   number(Lost),
  394        Status = lost(Lost)
  395    ;   Unload \== nil
  396    ->  Status = unload
  397    ).
  398wsid_status(WSID, Status) :-
  399    wsid_status_db(WSID, Status).
  400
  401wsid_status_del(WSID) :-
  402    redis_key(unload(WSID), Server, UnloadKey),
  403    !,
  404    redis_key(lost(WSID), Server, LostKey),
  405    redis(Server,
  406          [ del(UnloadKey),
  407            del(LostKey)
  408          ]).
  409wsid_status_del(WSID) :-
  410    retractall(wsid_status_db(WSID, _Status)).
  411
  412wsid_status_del_lost(WSID) :-
  413    redis_key(lost(WSID), Server, Key),
  414    !,
  415    redis(Server, del(Key)).
  416wsid_status_del_lost(WSID) :-
  417    retractall(wsid_status_db(WSID, lost(_))).
  418
  419wsid_status_set_lost(WSID, Time) :-
  420    redis_key(lost(WSID), Server, Key),
  421    !,
  422    redis(Server, set(Key, Time)).
  423wsid_status_set_lost(WSID, Time) :-
  424    assertz(wsid_status_db(WSID, lost(Time))).
  425
  426wsid_status_set_unload(WSID) :-
  427    redis_key(unload(WSID), Server, Key),
  428    !,
  429    redis(Server, set(Key, true)).
  430wsid_status_set_unload(WSID) :-
  431    assertz(wsid_status_db(WSID, unload)).

 wsid_status_del_unload(+WSID) is semidet

True when WSID has status unload and this status is removed.

  437wsid_status_del_unload(WSID) :-
  438    redis_key_ro(unload(WSID), ROServer, Key),
  439    !,
  440    (   redis(ROServer, get(Key), true)
  441    ->  redis_key(unload(WSID), RWServer, Key),
  442        redis(RWServer, del(Key))
  443    ).
  444wsid_status_del_unload(WSID) :-
  445    retract(wsid_status_db(WSID, unload)),
  446    !.

 register_wsid_session(+WSID, +Session) is det

Register WSID to belong to Session, i.e., the given websocket belongs to a Session on some browser. When using a Redis cluster, we also want to know the Redis consumer on which this session is active, such that we can relay messages to the proper SWISH server.

Redis data:

• wsid: set of WSID
• session:WSID to at(Consumer,Session)

  460register_wsid_session(WSID, Session) :-
  461    redis_key(wsid, Server, SetKey),
  462    redis_key(session(WSID), Server, SessionKey),
  463    !,
  464    redis_consumer(Consumer),
  465    redis(Server, sadd(SetKey, WSID)),
  466    redis(Server, set(SessionKey, at(Consumer,Session) as prolog)).
  467register_wsid_session(WSID, Session) :-
  468    assertz(wsid_session_db(WSID, Session)).

 wsid_session(?WSID, ?Session) is nondet

 wsid_session(?WSID, ?Session, -Consumer) is nondet

True when there is a known visitor WSID that is associated with the HTTP Session, uses Token for reconnecting and runs on a node identified by the Redis Consumer.

  477wsid_session(WSID, Session) :-
  478    wsid_session(WSID, Session, _Consumer).
  479
  480wsid_session(WSID, Session, Consumer) :-
  481    use_redis,
  482    !,
  483    (   nonvar(Session)
  484    ->  http_session_data(wsid(WSID,_Token), Session)
  485    ;   current_wsid(WSID),
  486        redis_key_ro(session(WSID), Server, SessionKey),
  487        redis(Server, get(SessionKey), at(Consumer,Session))
  488    ).
  489wsid_session(WSID, Session, single) :-
  490    wsid_session_db(WSID, Session).

 wsid_session_reclaim(+WSID, -Session) is semidet

True when WSID was connected to Session and now no longer is.

  496wsid_session_reclaim(WSID, Session) :-
  497    redis_key_ro(session(WSID), ROServer, SessionKey),
  498    redis_key(wsid, WRServer, SetKey),
  499    !,
  500    redis(ROServer, get(SessionKey), At),
  501    arg(2, At, Session),            % changed from at/3 to at/2.
  502    redis(WRServer, srem(SetKey, WSID)),
  503    redis(WRServer, del(SessionKey)).
  504wsid_session_reclaim(WSID, Session) :-
  505    retract(wsid_session_db(WSID, Session)).

 wsid_session_reclaim_all(+WSID, +Session) is det

  509wsid_session_reclaim_all(WSID, _Session) :-
  510    redis_key(wsid, Server, SetKey),
  511    !,
  512    redis(Server, srem(SetKey, WSID)),
  513    redis_key(session(WSID), Server, SessionKey),
  514    redis(Server, del(SessionKey)).
  515wsid_session_reclaim_all(WSID, Session) :-
  516    retractall(wsid_session_db(WSID, Session)).
  517
  518wsid_session_del_session(Session) :-
  519    use_redis,
  520    !,
  521    (   wsid_session(WSID, Session),
  522        wsid_session_reclaim(WSID, Session),
  523        fail
  524    ;   true
  525    ).
  526wsid_session_del_session(Session) :-
  527    retractall(wsid_session_db(_, Session)).

 current_wsid(?WSID) is nondet

True when WSID is a (Redis) known WSID.

  533current_wsid(WSID) :-
  534    nonvar(WSID),
  535    !,
  536    redis_key_ro(wsid, Server, SetKey),
  537    redis(Server, sismember(SetKey, WSID), 1).
  538current_wsid(WSID) :-
  539    redis_key_ro(wsid, Server, SetKey),
  540    redis_sscan(Server, SetKey, List, []),
  541    member(WSID, List).

 session_user(?Session, ?TmpUser:atom)

Relate Session to a tmp user id. Info about the tmp user is maintained in visitor_data/2.

  548session_user(Session, TmpUser) :-
  549    http_current_session(Session, swish_user(TmpUser)).
  550
  551session_user_create(Session, User) :-
  552    http_session_asserta(swish_user(User), Session).
  553
  554session_user_del(Session, User) :-
  555    http_session_retract(swish_user(User), Session).

 visitor_data(?Visitor, ?Data)

  559visitor_data(Visitor, Data) :-
  560    redis_key(visitor(Visitor), Server, Key),
  561    !,
  562    redis_get_hash(Server, Key, Data).
  563visitor_data(Visitor, Data) :-
  564    visitor_data_db(Visitor, Data).
  565
  566visitor_data_set(Visitor, Data) :-
  567    redis_key(visitor(Visitor), Server, Key),
  568    !,
  569    redis_set_hash(Server, Key, Data).
  570visitor_data_set(Visitor, Data) :-
  571    retractall(visitor_data_db(Visitor, _)),
  572    assertz(visitor_data_db(Visitor, Data)).
  573
  574visitor_data_del(Visitor, Data) :-
  575    redis_key(visitor(Visitor), Server, Key),
  576    !,
  577    redis_get_hash(Server, Key, Data),
  578    redis(Server, del(Key)).
  579visitor_data_del(Visitor, Data) :-
  580    retract(visitor_data_db(Visitor, Data)).

 subscription(?WSID, ?Channel, ?SubChannel)

Requires both WSID -> Channel/SubChannel and backward relation. Redis:

channel:SubChannel --> set(WSID-Channel) subscription:WSID --> set(Channel-SubChannel)

  590subscription(WSID, Channel, SubChannel) :-
  591    use_redis,
  592    !,
  593    (   nonvar(WSID), nonvar(Channel), nonvar(SubChannel)
  594    ->  redis_key_ro(subscription(WSID), Server, WsKey),
  595        redis(Server, sismember(WsKey, Channel-SubChannel as prolog), 1)
  596    ;   nonvar(SubChannel)
  597    ->  redis_key_ro(channel(SubChannel), Server, ChKey),
  598        redis_sscan(Server, ChKey, List, []),
  599        member(WSID-Channel, List)
  600    ;   (   nonvar(WSID)
  601        ->  true
  602        ;   current_wsid(WSID)
  603        ),
  604        redis_key_ro(subscription(WSID), Server, WsKey),
  605        redis_sscan(Server, WsKey, List, []),
  606        member(Channel-SubChannel, List)
  607    ).
  608subscription(WSID, Channel, SubChannel) :-
  609    subscription_db(WSID, Channel, SubChannel).

 subscribe(+WSID, +Channel) is det

 subscribe(+WSID, +Channel, +SubChannel) is det

Subscript WSID to listen to messages on Channel/SubChannel.

  616subscribe(WSID, Channel) :-
  617    subscribe(WSID, Channel, _SubChannel).
  618
  619subscribe(WSID, Channel, SubChannel) :-
  620    use_redis,
  621    !,
  622    redis_key(channel(SubChannel), Server, ChKey),
  623    redis_key(subscription(WSID), Server, WsKey),
  624    redis(Server, sadd(ChKey, WSID-Channel as prolog)),
  625    redis(Server, sadd(WsKey, Channel-SubChannel as prolog)).
  626subscribe(WSID, Channel, SubChannel) :-
  627    (   subscription(WSID, Channel, SubChannel)
  628    ->  true
  629    ;   assertz(subscription_db(WSID, Channel, SubChannel))
  630    ).

 unsubscribe(?WSID, ?Channel) is det

 unsubscribe(?WSID, ?Channel, ?SubChannel) is det

Remove all matching subscriptions.

  637unsubscribe(WSID, Channel) :-
  638    unsubscribe(WSID, Channel, _SubChannel).
  639
  640unsubscribe(WSID, Channel, SubChannel) :-
  641    use_redis,
  642    !,
  643    (   (   nonvar(WSID), nonvar(Channel), nonvar(SubChannel)
  644        ->  true
  645        ;   subscription(WSID, Channel, SubChannel)
  646        ),
  647        redis_unsubscribe(WSID, Channel, SubChannel),
  648        fail
  649    ;   true
  650    ).
  651unsubscribe(WSID, Channel, SubChannel) :-
  652    retractall(subscription_db(WSID, Channel, SubChannel)).
  653
  654redis_unsubscribe(WSID, Channel, SubChannel) :-
  655    redis_key(channel(SubChannel), Server, ChKey),
  656    redis_key(subscription(WSID), Server, WsKey),
  657    redis(Server, srem(ChKey, WSID-Channel as prolog)),
  658    redis(Server, srem(WsKey, Channel-SubChannel as prolog)).
  659
  660
  661		 /*******************************
  662		 *        HIGH LEVEL DB		*
  663		 *******************************/

 visitor(?WSID) is nondet

 visitor(?WSID, -Consumer) is nondet

True when WSID should be considered an active visitor that is connected to the SWISH node identified by the Redis Consumer. This means

• If we lost the visitor for less than 30 sec, we consider it still active.
• If it is connected to our websocket hub, it is active.
• Otherwise, if it runs on our node, we destroy the visitor.

  677visitor(WSID) :-
  678    visitor(WSID, _).
  679
  680visitor(WSID, Consumer) :-
  681    wsid_session(WSID, _Session, Consumer),
  682    \+ pending_visitor(WSID, 30).
  683
  684visitor_count(Count) :-
  685    use_redis,
  686    !,
  687    active_wsid_count(Count).
  688visitor_count(Count) :-
  689    aggregate_all(count, visitor(_), Count).

 pending_visitor(+WSID, +Timeout) is semidet

True if WSID is inactive. This means we lost the connection at least Timeout seconds ago.

  696pending_visitor(WSID, Timeout) :-
  697    wsid_status(WSID, lost(Lost)),
  698    get_time(Now),
  699    Now - Lost > Timeout.

 wsid_visitor(?WSID, ?Visitor)

True when WSID is associated with Visitor

  705wsid_visitor(WSID, Visitor) :-
  706    nonvar(WSID),
  707    !,
  708    wsid_session(WSID, Session),
  709    session_user(Session, Visitor).
  710wsid_visitor(WSID, Visitor) :-
  711    session_user(Session, Visitor),
  712    wsid_session(WSID, Session).

 existing_visitor(+WSID, +Session, -TmpUser, -UserData) is semidet

True if we are dealing with an existing visitor for which we lost the connection.

  719existing_visitor(WSID, Session, TmpUser, UserData) :-
  720    wsid_session(WSID, Session),
  721    session_user(Session, TmpUser),
  722    visitor_data(TmpUser, UserData),
  723    !.
  724existing_visitor(WSID, Session, _, _) :-
  725    wsid_session_reclaim_all(WSID, Session),
  726    fail.

 create_visitor(+WSID, +Session, -TmpUser, -UserData, +Options)

Create a new visitor when a new websocket is established. Options provides information we have about the user:

current_user_info(+Info)

Already logged in user with given information

avatar(Avatar)

Avatar remembered in the browser for this user.

anonymous_name(NickName)

Nick name remembered in the browser for this user.

anonymous_avatar(URL)

Avatar remembered in the browser for this user. Using the SVG avatars, this is `/icons/avatar.svg#NNN`, which NNN is a bitmask on the SVG to change its appearance,

  744create_visitor(WSID, Session, TmpUser, UserData, Options) :-
  745    register_wsid_session(WSID, Session),
  746    create_session_user(Session, TmpUser, UserData, Options).

 random_key(+Len, -Key) is det

Generate a random confirmation key

  752random_key(Len, Key) :-
  753    length(Codes, Len),
  754    maplist(random_between(0,255), Codes),
  755    phrase(base64url(Codes), Encoded),
  756    atom_codes(Key, Encoded).

 destroy_visitor(+WSID) is det

The web socket WSID has been closed. We should not immediately destroy the temporary user as the browser may soon reconnect due to a page reload or re-establishing the web socket after a temporary network failure. We leave the destruction thereof to the session, but set the session timeout to a fairly short time.

To be done

- We should only inform clients that we have informed about this user.

  769destroy_visitor(WSID) :-
  770    must_be(atom, WSID),
  771    update_session_timeout(WSID),
  772    destroy_reason(WSID, Reason),
  773    (   Reason == unload
  774    ->  reclaim_visitor(WSID)
  775    ;   get_time(Now),
  776        wsid_status_set_lost(WSID, Now)
  777    ),
  778    visitor_count(Count),
  779    debug(chat(visitor), '~p left. Broadcasting ~d visitors', [WSID,Count]),
  780    chat_broadcast(_{ type:removeUser,
  781                      wsid:WSID,
  782                      reason:Reason,
  783                      visitors:Count
  784                    }).
  785
  786destroy_reason(WSID, Reason) :-
  787    wsid_status_del_unload(WSID),
  788    !,
  789    Reason = unload.
  790destroy_reason(_, close).

 update_session_timeout(+WSID) is det

WSID was lost. If this is the only websocket on this session we reduce the session timeout to 15 minutes. This cooperates with accept_chat/3, which sets the session timeout to 100 days for as long as we have an active websocket connection.

  799update_session_timeout(WSID) :-
  800    wsid_session(WSID, Session),
  801    !,
  802    (   wsid_session(WSID2, Session),
  803        WSID2 \== WSID
  804    ->  true
  805    ;   debug(chat(websocket), 'Websocket ~p was last in session ~p',
  806              [ WSID, Session]),
  807        http_set_session(Session, timeout(900))
  808    ).
  809update_session_timeout(_).

 gc_visitors

Reclaim all visitors with whom we have lost the connection and the browser did not reclaim the session within session_lost_timeout seconds.

This also updates active_wsid/2 to reflect the current status.

  819:- dynamic gc_status/1.  820
  821gc_visitors :-
  822    swish_config(session_lost_timeout, TMO),
  823    gc_status(Status),
  824    (   Status == running
  825    ->  true
  826    ;   Status = completed(When),
  827        get_time(Now),
  828        Now-When > TMO
  829    ->  fail
  830    ;   retractall(gc_status(completed(_)))
  831    ),
  832    !.
  833gc_visitors :-
  834    swish_config(session_lost_timeout, TMO),
  835    catch(thread_create(gc_visitors_sync(TMO), _Id,
  836                        [ alias('swish_chat_gc_visitors'),
  837                          detached(true)
  838                        ]),
  839          error(permission_error(create, thread, _), _),
  840          true).
  841
  842gc_visitors_sync(TMO) :-
  843    setup_call_cleanup(
  844        asserta(gc_status(running), Ref),
  845        ( do_gc_visitors(TMO),
  846          get_time(Now),
  847          asserta(gc_status(completed(Now)))
  848        ),
  849        erase(Ref)).
  850
  851do_gc_visitors(TMO) :-
  852    findall(WSID-Consumer,
  853            active_visitor(TMO, WSID, Consumer),
  854            Pairs),
  855    transaction(
  856        ( retractall(active_wsid(_,_)),
  857          forall(member(WSID-Consumer, Pairs),
  858                 assertz(active_wsid(WSID, Consumer))))).
  859
  860active_visitor(TMO, WSID, Consumer) :-
  861    wsid_session(WSID, _Session, Consumer),
  862    (   valid_visitor(WSID, TMO, Consumer)
  863    ->  true
  864    ;   reclaim_visitor(WSID),
  865        fail
  866    ).
  867
  868valid_visitor(WSID, _TMO, _Consumer) :-
  869    hub_member(swish_chat, WSID),
  870    !.
  871valid_visitor(WSID, TMO, _Consumer) :-
  872    wsid_status(WSID, lost(Lost)),
  873    !,
  874    get_time(Now),
  875    Now - Lost < TMO.
  876valid_visitor(_WSID, _TMO, Consumer) :-
  877    use_redis,
  878    !,
  879    \+ redis_consumer(Consumer).

 reclaim_visitor(+WSID) is det

Reclaim a WSID connection. If the user left gracefully, this is called immediately. If we lost the connection on an error, this is eventually called (indirectly) by do_gc_visitors/1.

  887reclaim_visitor(WSID) :-
  888    debug(chat(gc), 'Reclaiming idle ~p', [WSID]),
  889    reclaim_wsid_session(WSID),
  890    wsid_status_del(WSID),
  891    unsubscribe(WSID, _).
  892
  893reclaim_wsid_session(WSID) :-
  894    (   wsid_session_reclaim(WSID, Session)
  895    ->  http_session_retractall(websocket(_, _), Session)
  896    ;   true
  897    ).

 create_session_user(+Session, -User, -UserData, +Options)

Associate a user with the session. The user id is a UUID that is not associated with any persistent notion of a user. The destruction is left to the destruction of the session.

  905:- listen(http_session(end(SessionID, _Peer)),
  906          destroy_session_user(SessionID)).  907
  908create_session_user(Session, TmpUser, UserData, _Options) :-
  909    session_user(Session, TmpUser),
  910    visitor_data(TmpUser, UserData),
  911    !.
  912create_session_user(Session, TmpUser, UserData, Options) :-
  913    uuid(TmpUser),
  914    get_visitor_data(UserData, Options),
  915    session_user_create(Session, TmpUser),
  916    visitor_data_set(TmpUser, UserData).
  917
  918destroy_session_user(Session) :-
  919    forall(wsid_session(WSID, Session, _Token),
  920           inform_session_closed(WSID, Session)),
  921    wsid_session_del_session(Session),
  922    forall(session_user_del(Session, TmpUser),
  923           destroy_visitor_data(TmpUser)).
  924
  925destroy_visitor_data(TmpUser) :-
  926    (   visitor_data_del(TmpUser, Data),
  927        release_avatar(Data.get(avatar)),
  928        fail
  929    ;   true
  930    ).
  931
  932inform_session_closed(WSID, Session) :-
  933    ignore(hub_send(WSID, json(_{type:session_closed}))),
  934    session_user(Session, TmpUser),
  935    update_visitor_data(TmpUser, _Data, logout).

 update_visitor_data(+TmpUser, +Data, +Reason) is det

Update the user data for the visitor TmpUser to Data. This is rather complicated due to all the defaulting rules. Reason is one of:

• login
• logout
• 'set-nick-name'
• 'profile-edit'

To be done

- Create a more declarative description on where the various attributes must come from.

  952update_visitor_data(TmpUser, _Data, logout) :-
  953    !,
  954    anonymise_user_data(TmpUser, NewData),
  955    set_visitor_data(TmpUser, NewData, logout).
  956update_visitor_data(TmpUser, Data, Reason) :-
  957    profile_reason(Reason),
  958    !,
  959    (   visitor_data(TmpUser, Old)
  960    ;   Old = v{}
  961    ),
  962    copy_profile([name,avatar,email], Data, Old, New),
  963    set_visitor_data(TmpUser, New, Reason).
  964update_visitor_data(TmpUser, _{name:Name}, 'set-nick-name') :-
  965    !,
  966    visitor_data(TmpUser, Old),
  967    set_nick_name(Old, Name, New),
  968    set_visitor_data(TmpUser, New, 'set-nick-name').
  969update_visitor_data(TmpUser, Data, Reason) :-
  970    set_visitor_data(TmpUser, Data, Reason).
  971
  972profile_reason('profile-edit').
  973profile_reason('login').
  974
  975copy_profile([], _, Data, Data).
  976copy_profile([H|T], New, Data0, Data) :-
  977    copy_profile_field(H, New, Data0, Data1),
  978    copy_profile(T, New, Data1, Data).
  979
  980copy_profile_field(avatar, New, Data0, Data) :-
  981    !,
  982    (   Data1 = Data0.put(avatar,New.get(avatar))
  983    ->  Data  = Data1.put(avatar_source, profile)
  984    ;   email_gravatar(New.get(email), Avatar),
  985        valid_gravatar(Avatar)
  986    ->  Data = Data0.put(_{avatar:Avatar,avatar_source:email})
  987    ;   Avatar = Data0.get(anonymous_avatar)
  988    ->  Data = Data0.put(_{avatar:Avatar,avatar_source:client})
  989    ;   noble_avatar_url(Avatar, []),
  990        Data = Data0.put(_{avatar:Avatar,avatar_source:generated,
  991                           anonymous_avatar:Avatar
  992                          })
  993    ).
  994copy_profile_field(email, New, Data0, Data) :-
  995    !,
  996    (   NewMail = New.get(email)
  997    ->  update_avatar_from_email(NewMail, Data0, Data1),
  998        Data = Data1.put(email, NewMail)
  999    ;   update_avatar_from_email('', Data0, Data1),
 1000        (   del_dict(email, Data1, _, Data)
 1001        ->  true
 1002        ;   Data = Data1
 1003        )
 1004    ).
 1005copy_profile_field(F, New, Data0, Data) :-
 1006    (   Data = Data0.put(F, New.get(F))
 1007    ->  true
 1008    ;   del_dict(F, Data0, _, Data)
 1009    ->  true
 1010    ;   Data = Data0
 1011    ).
 1012
 1013set_nick_name(Data0, Name, Data) :-
 1014    Data = Data0.put(_{name:Name, anonymous_name:Name}).

 update_avatar_from_email(+Email, +DataIn, -Data)

Update the avatar after a change of the known email. If the avatar comes from the profile, no action is needed. If Email has a gravatar, use that. Else use the know or a new generated avatar.

 1023update_avatar_from_email(_, Data, Data) :-
 1024    Data.get(avatar_source) == profile,
 1025    !.
 1026update_avatar_from_email('', Data0, Data) :-
 1027    Data0.get(avatar_source) == email,
 1028    !,
 1029    noble_avatar_url(Avatar, []),
 1030    Data = Data0.put(_{avatar:Avatar, anonymous_avatar:Avatar,
 1031                       avatar_source:generated}).
 1032update_avatar_from_email(Email, Data0, Data) :-
 1033    email_gravatar(Email, Avatar),
 1034    valid_gravatar(Avatar),
 1035    !,
 1036    Data = Data0.put(avatar, Avatar).
 1037update_avatar_from_email(_, Data0, Data) :-
 1038    (   Avatar = Data0.get(anonymous_avatar)
 1039    ->  Data = Data0.put(_{avatar:Avatar, avatar_source:client})
 1040    ;   noble_avatar_url(Avatar, []),
 1041        Data = Data0.put(_{avatar:Avatar, anonymous_avatar:Avatar,
 1042                           avatar_source:generated})
 1043    ).

 anonymise_user_data(TmpUser, Data)

Create anonymous user profile.

 1049anonymise_user_data(TmpUser, Data) :-
 1050    visitor_data(TmpUser, Old),
 1051    (   _{anonymous_name:AName, anonymous_avatar:AAvatar} :< Old
 1052    ->  Data = _{anonymous_name:AName, anonymous_avatar:AAvatar,
 1053                 name:AName, avatar:AAvatar, avatar_source:client}
 1054    ;   _{anonymous_avatar:AAvatar} :< Old
 1055    ->  Data = _{anonymous_avatar:AAvatar,
 1056                 avatar:AAvatar, avatar_source:client}
 1057    ;   _{anonymous_name:AName} :< Old
 1058    ->  noble_avatar_url(Avatar, []),
 1059        Data = _{anonymous_name:AName, anonymous_avatar:Avatar,
 1060                 name:AName, avatar:Avatar, avatar_source:generated}
 1061    ),
 1062    !.
 1063anonymise_user_data(_, Data) :-
 1064    noble_avatar_url(Avatar, []),
 1065    Data = _{anonymous_avatar:Avatar,
 1066             avatar:Avatar, avatar_source:generated}.

 set_visitor_data(+TmpUser, +Data, +Reason) is det

Update the user data for the session user TmpUser and forward the changes.

 1073set_visitor_data(TmpUser, Data, Reason) :-
 1074    visitor_data_set(TmpUser, Data),
 1075    inform_visitor_change(TmpUser, Reason).

 inform_visitor_change(+TmpUser, +Reason) is det

Inform browsers showing TmpUser that the visitor data has changed. The first clause deals with forwarding from HTTP requests, where we have the session and the second from websocket requests where we have the WSID.

 1084inform_visitor_change(TmpUser, Reason) :-
 1085    http_in_session(Session),
 1086    !,
 1087    public_user_data(TmpUser, Data),
 1088    forall(wsid_session(WSID, Session),
 1089           inform_friend_change(WSID, Data, Reason)).
 1090inform_visitor_change(TmpUser, Reason) :-
 1091    nb_current(wsid, WSID),
 1092    !,
 1093    public_user_data(TmpUser, Data),
 1094    inform_friend_change(WSID, Data, Reason).
 1095inform_visitor_change(_, _).
 1096
 1097inform_friend_change(WSID, Data, Reason) :-
 1098    Message = json(_{ type:"profile",
 1099                      wsid:WSID,
 1100                      reason:Reason
 1101                    }.put(Data)),
 1102    send_friends(WSID, Message).

 sync_gazers(+WSID, +Files:list(atom)) is det

A browser signals it has Files open. This happens when a SWISH instance is created as well as when a SWISH instance changes state, such as closing a tab, adding a tab, bringing a tab to the foreground, etc.

 1111sync_gazers(WSID, Files0) :-
 1112    findall(F, subscription(WSID, gitty, F), Viewing0),
 1113    sort(Files0, Files),
 1114    sort(Viewing0, Viewing),
 1115    (   Files == Viewing
 1116    ->  true
 1117    ;   ord_subtract(Files, Viewing, New),
 1118        add_gazing(WSID, New),
 1119        ord_subtract(Viewing, Files, Left),
 1120        del_gazing(WSID, Left)
 1121    ).
 1122
 1123add_gazing(_, []) :- !.
 1124add_gazing(WSID, Files) :-
 1125    inform_me_about_existing_gazers(WSID, Files),
 1126    inform_existing_gazers_about_newby(WSID, Files).
 1127
 1128inform_me_about_existing_gazers(WSID, Files) :-
 1129    hub_member(swish_chat, WSID),
 1130    !,
 1131    findall(Gazer, files_gazer(Files, Gazer), Gazers),
 1132    ignore(hub_send(WSID, json(_{type:"gazers", gazers:Gazers}))).
 1133inform_me_about_existing_gazers(_, _).
 1134
 1135files_gazer(Files, Gazer) :-
 1136    member(File, Files),
 1137    subscription(WSID, gitty, File),
 1138    wsid_session(WSID, Session),
 1139    session_user(Session, UID),
 1140    public_user_data(UID, Data),
 1141    Gazer = _{file:File, uid:UID, wsid:WSID}.put(Data).
 1142
 1143inform_existing_gazers_about_newby(WSID, Files) :-
 1144    forall(member(File, Files),
 1145           signal_gazer(WSID, File)).
 1146
 1147signal_gazer(WSID, File) :-
 1148    subscribe(WSID, gitty, File),
 1149    broadcast_event(opened(File), File, WSID).
 1150
 1151del_gazing(_, []) :- !.
 1152del_gazing(WSID, Files) :-
 1153    forall(member(File, Files),
 1154           del_gazing1(WSID, File)).
 1155
 1156del_gazing1(WSID, File) :-
 1157    broadcast_event(closed(File), File, WSID),
 1158    unsubscribe(WSID, gitty, File).

 add_user_details(+Message, -Enriched) is det

Add additional information to a message. Message must contain a uid field.

 1165add_user_details(Message, Enriched) :-
 1166    public_user_data(Message.uid, Data),
 1167    Enriched = Message.put(Data).

 public_user_data(+UID, -Public:dict) is det

True when Public provides the information we publically share about UID. This is currently the name and avatar.

 1174public_user_data(UID, Public) :-
 1175    visitor_data(UID, Data),
 1176    (   _{name:Name, avatar:Avatar} :< Data
 1177    ->  Public = _{name:Name, avatar:Avatar}
 1178    ;   _{avatar:Avatar} :< Data
 1179    ->  Public = _{avatar:Avatar}
 1180    ;   Public = _{}
 1181    ).

 get_visitor_data(-Data:dict, +Options) is det

Optain data for a new visitor. Options include:

identity(+Identity)

Identity information provided by authenticate/2. Always present.

avatar(+URL)

Avatar provided by the user

nick_name(+Name)

Nick name provided by the user.

Data always contains an avatar key and optionally contains a name and email key. If the avatar is generated there is also a key avatar_generated with the value true.

bug

- This may check for avatar validity, which may take long. Possibly we should do this in a thread.

 1202get_visitor_data(Data, Options) :-
 1203    option(identity(Identity), Options),
 1204    findall(N-V, visitor_property(Identity, Options, N, V), Pairs),
 1205    dict_pairs(Data, v, Pairs).
 1206
 1207visitor_property(Identity, Options, name, Name) :-
 1208    (   user_property(Identity, name(Name))
 1209    ->  true
 1210    ;   option(nick_name(Name), Options)
 1211    ).
 1212visitor_property(Identity, _, email, Email) :-
 1213    user_property(Identity, email(Email)).
 1214visitor_property(Identity, Options, Name, Value) :-
 1215    (   user_property(Identity, avatar(Avatar))
 1216    ->  avatar_property(Avatar, profile, Name, Value)
 1217    ;   user_property(Identity, email(Email)),
 1218        email_gravatar(Email, Avatar),
 1219        valid_gravatar(Avatar)
 1220    ->  avatar_property(Avatar, email, Name, Value)
 1221    ;   option(avatar(Avatar), Options)
 1222    ->  avatar_property(Avatar, client, Name, Value)
 1223    ;   noble_avatar_url(Avatar, Options),
 1224        avatar_property(Avatar, generated, Name, Value)
 1225    ).
 1226visitor_property(_, Options, anonymous_name, Name) :-
 1227    option(nick_name(Name), Options).
 1228visitor_property(_, Options, anonymous_avatar, Avatar) :-
 1229    option(avatar(Avatar), Options).
 1230
 1231
 1232avatar_property(Avatar, _Source, avatar,        Avatar).
 1233avatar_property(_Avatar, Source, avatar_source, Source).
 1234
 1235
 1236                 /*******************************
 1237                 *         NOBLE AVATAR         *
 1238                 *******************************/
 1239
 1240:- http_handler(swish('avatar/'), reply_avatar, [id(avatar), prefix]).

 reply_avatar(+Request)

HTTP handler for Noble Avatar images. Using create_avatar/2 re-creates avatars from the file name, so we can safely discard the avatar file store.

Not really. A new user gets a new avatar and this is based on whether or not the file exists. Probably we should maintain a db of handed out avatars and their last-use time stamp. How to do that? Current swish stats: 400K avatars, 3.2Gb data.

 1253reply_avatar(Request) :-
 1254    cors_enable,
 1255    option(path_info(Local), Request),
 1256    (   absolute_file_name(noble_avatar(Local), Path,
 1257                           [ access(read),
 1258                             file_errors(fail)
 1259                           ])
 1260    ->  true
 1261    ;   create_avatar(Local, Path)
 1262    ),
 1263    http_reply_file(Path, [unsafe(true)], Request).
 1264
 1265
 1266noble_avatar_url(HREF, Options) :-
 1267    option(avatar(HREF), Options),
 1268    !.
 1269noble_avatar_url(HREF, _Options) :-
 1270    swish_config:config(avatars, noble),
 1271    !,
 1272    noble_avatar(_Gender, Path, true),
 1273    file_base_name(Path, File),
 1274    http_absolute_location(swish(avatar/File), HREF, []).
 1275noble_avatar_url(HREF, _Options) :-
 1276    A is random(0x1FFFFF+1),
 1277    http_absolute_location(icons('avatar.svg'), HREF0, []),
 1278    format(atom(HREF), '~w#~d', [HREF0, A]).
 1279
 1280
 1281
 1282                 /*******************************
 1283                 *         BROADCASTING         *
 1284                 *******************************/

 chat_broadcast(+Message) is det

 chat_broadcast(+Message, +Channel) is det

Send Message to all known SWISH clients. Message is a valid JSON object, i.e., a dict or option list. When using Redis we send the message to the swish:chat pubsub channel and listening for swish:chat calls chat_broadcast_local/1,2 in each instance.

Arguments:

Channel

- is either an atom or a term Channel/SubChannel, where both Channel and SubChannel are atoms.

 1297chat_broadcast(Message) :-
 1298    use_redis,
 1299    !,
 1300    redis(swish, publish(swish:chat, chat(Message) as prolog)).
 1301chat_broadcast(Message) :-
 1302    chat_broadcast_local(Message).
 1303
 1304chat_broadcast(Message, Channel) :-
 1305    use_redis,
 1306    !,
 1307    redis(swish, publish(swish:chat, chat(Message, Channel) as prolog)).
 1308chat_broadcast(Message, Channel) :-
 1309    chat_broadcast_local(Message, Channel).
 1310
 1311
 1312chat_broadcast_local(Message) :-
 1313    debug(chat(broadcast), 'Broadcast: ~p', [Message]),
 1314    hub_broadcast(swish_chat, json(Message)).
 1315
 1316chat_broadcast_local(Message, Channel/SubChannel) :-
 1317    !,
 1318    must_be(atom, Channel),
 1319    must_be(atom, SubChannel),
 1320    debug(chat(broadcast), 'Broadcast on ~p: ~p',
 1321          [Channel/SubChannel, Message]),
 1322    hub_broadcast(swish_chat, json(Message),
 1323                  subscribed(Channel, SubChannel)).
 1324chat_broadcast_local(Message, Channel) :-
 1325    must_be(atom, Channel),
 1326    debug(chat(broadcast), 'Broadcast on ~p: ~p', [Channel, Message]),
 1327    hub_broadcast(swish_chat, json(Message),
 1328                  subscribed(Channel)).

 subscribed(+Channel, +WSID) is semidet

 subscribed(+Channel, +SubChannel, +WSID) is semidet

Filter used by hub_broadcast/3. WSID is always a locally known web and active socket.

 1336subscribed(Channel, WSID) :-
 1337    subscription(WSID, Channel, _).
 1338subscribed(Channel, SubChannel, WSID) :-
 1339    subscription(WSID, Channel, SubChannel).
 1340subscribed(gitty, SubChannel, WSID) :-
 1341    swish_config:config(hangout, SubChannel),
 1342    \+ subscription(WSID, gitty, SubChannel).

 send_friends(+WSID, +Message)

Send Message to WSID and all its friends.

 1348send_friends(WSID, Message) :-
 1349    use_redis,
 1350    !,
 1351    redis(swish, publish(swish:chat, send_friends(WSID, Message) as prolog)).
 1352send_friends(WSID, Message) :-
 1353    send_friends_local(WSID, Message).
 1354
 1355send_friends_local(WSID, Message) :-
 1356    hub_send_if_on_me(WSID, Message),
 1357    forall(distinct(viewing_same_file(WSID, Friend)),
 1358           ignore(hub_send_if_on_me(Friend, Message))).
 1359
 1360hub_send_if_on_me(WSID, Message) :-
 1361    hub_member(swish_chat, WSID),
 1362    !,
 1363    hub_send(WSID, Message).
 1364hub_send_if_on_me(_, _).
 1365
 1366viewing_same_file(WSID, Friend) :-
 1367    subscription(WSID, gitty, File),
 1368    subscription(Friend, gitty, File),
 1369    Friend \== WSID.
 1370
 1371
 1372		 /*******************************
 1373		 *      REDIS CONNNECTION       *
 1374		 *******************************/
 1375
 1376:- initialization
 1377    listen(redis(_, 'swish:chat', Message),
 1378           chat_message(Message)). 1379
 1380chat_message(chat(Message)) :-
 1381    update_visitors(Message),
 1382    chat_broadcast_local(Message).
 1383chat_message(chat(Message, Channel)) :-
 1384    chat_broadcast_local(Message, Channel).
 1385chat_message(send_friends(WSID, Message)) :-
 1386    send_friends_local(WSID, Message).

 update_visitors(+Msg) is det

Maintain notion of active users based on broadcasted (re)join and left messages. We sync every 5 minutes to compensate for possible missed users.

 1394:- dynamic
 1395       (   last_wsid_sync/1,
 1396	   active_wsid/2
 1397       ) as volatile. 1398
 1399update_visitors(Msg),
 1400  _{type:removeUser, wsid:WSID} :< Msg =>
 1401    retractall(active_wsid(WSID, _)).
 1402update_visitors(Msg),
 1403  _{type:joined, wsid:WSID} :< Msg,
 1404  \+ active_wsid(WSID, _) =>
 1405    asserta(active_wsid(WSID, Msg.get(consumer, -))).
 1406update_visitors(Msg),
 1407  _{type:rejoined, wsid:WSID} :< Msg,
 1408  \+ active_wsid(WSID, _) =>
 1409    asserta(active_wsid(WSID, Msg.get(consumer, -))).
 1410update_visitors(_) =>
 1411    true.
 1412
 1413active_wsid_count(Count) :-
 1414    predicate_property(active_wsid(_,_), number_of_clauses(Count)),
 1415    !.
 1416active_wsid_count(0).
 1417
 1418active_wsid_count(Consumer, Count) :-
 1419    aggregate(count, WSID^active_wsid(WSID, Consumer), Count).
 1420
 1421
 1422                 /*******************************
 1423                 *           CHAT ROOM          *
 1424                 *******************************/
 1425
 1426create_chat_room :-
 1427    current_hub(swish_chat, _),
 1428    !.
 1429create_chat_room :-
 1430    with_mutex(swish_chat, create_chat_room_sync).
 1431
 1432create_chat_room_sync :-
 1433    current_hub(swish_chat, _),
 1434    !.
 1435create_chat_room_sync :-
 1436    hub_create(swish_chat, Room, _{}),
 1437    thread_create(swish_chat(Room), _, [alias(swish_chat)]).
 1438
 1439swish_chat(Room) :-
 1440    (   catch_with_backtrace(swish_chat_event(Room), E, chat_exception(E))
 1441    ->  true
 1442    ;   print_message(warning, goal_failed(swish_chat_event(Room)))
 1443    ),
 1444    swish_chat(Room).
 1445
 1446chat_exception('$aborted') :- !.
 1447chat_exception(E) :-
 1448    print_message(warning, E).
 1449
 1450swish_chat_event(Room) :-
 1451    thread_get_message(Room.queues.event, Message),
 1452    (   handle_message(Message, Room)
 1453    ->  true
 1454    ;   print_message(warning, goal_failed(handle_message(Message, Room)))
 1455    ).

 handle_message(+Message, +Room)

Handle incoming messages. This handles messages from our websocket connections, i.e., this does not see messages on other (Redis) instances.

 1463handle_message(Message, _Room) :-
 1464    websocket{opcode:text} :< Message,
 1465    !,
 1466    atom_json_dict(Message.data, JSON, []),
 1467    debug(chat(received), 'Received from ~p: ~p', [Message.client, JSON]),
 1468    WSID = Message.client,
 1469    (   current_wsid(WSID)
 1470    ->  setup_call_cleanup(
 1471            b_setval(wsid, WSID),
 1472            json_message(JSON, WSID),
 1473            nb_delete(wsid))
 1474    ;   debug(chat(visitor), 'Ignored ~p (WSID ~p unknown)', [Message, WSID])
 1475    ).
 1476handle_message(Message, _Room) :-
 1477    hub{joined:WSID} :< Message,
 1478    !,
 1479    debug(chat(visitor), 'Joined: ~p', [WSID]).
 1480handle_message(Message, _Room) :-
 1481    hub{left:WSID, reason:write(Lost)} :< Message,
 1482    !,
 1483    (   destroy_visitor(WSID)
 1484    ->  debug(chat(visitor), 'Left ~p due to write error for ~p',
 1485              [WSID, Lost])
 1486    ;   true
 1487    ).
 1488handle_message(Message, _Room) :-
 1489    hub{left:WSID} :< Message,
 1490    !,
 1491    (   destroy_visitor(WSID)
 1492    ->  debug(chat(visitor), 'Left: ~p', [WSID])
 1493    ;   true
 1494    ).
 1495handle_message(Message, _Room) :-
 1496    websocket{opcode:close, client:WSID} :< Message,
 1497    !,
 1498    debug(chat(visitor), 'Left: ~p', [WSID]),
 1499    destroy_visitor(WSID).
 1500handle_message(Message, _Room) :-
 1501    debug(chat(ignored), 'Ignoring chat message ~p', [Message]).

 json_message(+Message, +WSID) is det

Process a JSON message translated to a dict. The following messages are understood:

• subscribe channel [subchannel]
• unsubscribe channel [subchannel] Actively (un)subscribe for specific message channels.
• unload A SWISH instance is cleanly being unloaded.
• has-open-files files Executed after initiating the websocket to indicate loaded files.
• set-nick-name name User set nick name for anonymous identoty

 1520json_message(Dict, WSID) :-
 1521    _{ type: "subscribe",
 1522       channel:ChannelS, sub_channel:SubChannelS} :< Dict,
 1523    !,
 1524    atom_string(Channel, ChannelS),
 1525    atom_string(SubChannel, SubChannelS),
 1526    subscribe(WSID, Channel, SubChannel).
 1527json_message(Dict, WSID) :-
 1528    _{type: "subscribe", channel:ChannelS} :< Dict,
 1529    !,
 1530    atom_string(Channel, ChannelS),
 1531    subscribe(WSID, Channel).
 1532json_message(Dict, WSID) :-
 1533    _{ type: "unsubscribe",
 1534       channel:ChannelS, sub_channel:SubChannelS} :< Dict,
 1535    !,
 1536    atom_string(Channel, ChannelS),
 1537    atom_string(SubChannel, SubChannelS),
 1538    unsubscribe(WSID, Channel, SubChannel).
 1539json_message(Dict, WSID) :-
 1540    _{type: "unsubscribe", channel:ChannelS} :< Dict,
 1541    !,
 1542    atom_string(Channel, ChannelS),
 1543    unsubscribe(WSID, Channel).
 1544json_message(Dict, WSID) :-
 1545    _{type: "unload"} :< Dict,     % clean close/reload
 1546    !,
 1547    sync_gazers(WSID, []),
 1548    wsid_status_set_unload(WSID).
 1549json_message(Dict, WSID) :-
 1550    _{type: "has-open-files", files:FileDicts} :< Dict,
 1551    !,
 1552    maplist(dict_file_name, FileDicts, Files),
 1553    sync_gazers(WSID, Files).
 1554json_message(Dict, WSID) :-
 1555    _{type: "reloaded", file:FileS, commit:Hash} :< Dict,
 1556    !,
 1557    atom_string(File, FileS),
 1558    event_html(reloaded(File), HTML),
 1559    Message = _{ type:notify,
 1560                 wsid:WSID,
 1561                 html:HTML,
 1562                 event:reloaded,
 1563                 argv:[File,Hash]
 1564               },
 1565    chat_broadcast(Message, gitty/File).
 1566json_message(Dict, WSID) :-
 1567    _{type: "set-nick-name", name:Name} :< Dict,
 1568    !,
 1569    wsid_visitor(WSID, Visitor),
 1570    update_visitor_data(Visitor, _{name:Name}, 'set-nick-name').
 1571json_message(Dict, WSID) :-
 1572    _{type: "chat-message", docid:DocID} :< Dict,
 1573    !,
 1574    chat_add_user_id(WSID, Dict, Message),
 1575    (   forbidden(Message, DocID, Why)
 1576    ->  hub_send(WSID, json(json{type:forbidden,
 1577                                 action:chat_post,
 1578                                 about:DocID,
 1579                                 message:Why
 1580                                }))
 1581    ;   chat_relay(Message)
 1582    ).
 1583json_message(Dict, _WSID) :-
 1584    debug(chat(ignored), 'Ignoring JSON message ~p', [Dict]).
 1585
 1586dict_file_name(Dict, File) :-
 1587    atom_string(File, Dict.get(file)).

 forbidden(+Message, +DocID, -Why) is semidet

True if the chat Message about DocID must be forbidden, in which case Why is unified with a string indicating the reason. Currently:

• Demands the user to be logged on
• Limits the size of the message and its payloads

To be done

- Call authorized/2 with all proper identity information.

 1600forbidden(Message, DocID, Why) :-
 1601    \+ swish_config:config(chat_spam_protection, false),
 1602    \+ ws_authorized(chat(post(Message, DocID)), Message.user),
 1603    !,
 1604    Why = "Due to frequent spamming we were forced to limit \c
 1605               posting chat messages to users who are logged in.".
 1606forbidden(Message, _DocID, Why) :-
 1607    Text = Message.get(text),
 1608    string_length(Text, Len),
 1609    Len > 500,
 1610    Why = "Chat messages are limited to 500 characters".
 1611forbidden(Message, _DocID, Why) :-
 1612    Payloads = Message.get(payload),
 1613    member(Payload, Payloads),
 1614    large_payload(Payload, Why),
 1615    !.
 1616forbidden(Message, _DocID, Why) :-
 1617    \+ swish_config:config(chat_spam_protection, false),
 1618    eval_content(Message.get(text), _WC, Score),
 1619    user_score(Message, Score, Cummulative, _Count),
 1620    Score*2 + Cummulative < 0,
 1621    !,
 1622    Why = "Chat messages must be in English and avoid offensive language".
 1623
 1624large_payload(Payload, Why) :-
 1625    Selections = Payload.get(selection),
 1626    member(Selection, Selections),
 1627    (   string_length(Selection.get(string), SelLen), SelLen > 500
 1628    ;   string_length(Selection.get(context), SelLen), SelLen > 500
 1629    ),
 1630    !,
 1631    Why = "Selection too long (max. 500 characters)".
 1632large_payload(Payload, Why) :-
 1633    string_length(Payload.get(query), QLen), QLen > 1000,
 1634    !,
 1635    Why = "Query too long (max. 1000 characters)".
 1636
 1637user_score(Message, MsgScore, Cummulative, Count) :-
 1638    Profile = Message.get(user).get(profile_id),
 1639    !,
 1640    block(Profile, MsgScore, Cummulative, Count).
 1641user_score(_, _, 0, 1).

 block(+User, +Score, -Cummulative, -Count)

Keep a count and cummulative score for a user.

 1647:- dynamic
 1648    blocked/4. 1649
 1650block(User, Score, Cummulative, Count) :-
 1651    blocked(User, Score0, Count0, Time),
 1652    !,
 1653    get_time(Now),
 1654    Cummulative = Score0*(0.5**((Now-Time)/600)) + Score,
 1655    Count is Count0 + 1,
 1656    asserta(blocked(User, Cummulative, Count, Now)),
 1657    retractall(blocked(User, Score0, Count0, Time)).
 1658block(User, Score, Score, 1) :-
 1659    get_time(Now),
 1660    asserta(blocked(User, Score, 1, Now)).
 1661
 1662
 1663                 /*******************************
 1664                 *         CHAT MESSAGES        *
 1665                 *******************************/

 chat_add_user_id(+WSID, +Message0, -Message) is det

Decorate a message with the user credentials.

 1671chat_add_user_id(WSID, Dict, Message) :-
 1672    wsid_session(WSID, Session),
 1673    session_user(Session, Visitor),
 1674    visitor_data(Visitor, UserData),
 1675    User0 = u{avatar:UserData.avatar,
 1676              wsid:WSID
 1677             },
 1678    (   Name = UserData.get(name)
 1679    ->  User1 = User0.put(name, Name)
 1680    ;   User1 = User0
 1681    ),
 1682    (   http_current_session(Session, profile_id(ProfileID))
 1683    ->  User = User1.put(profile_id, ProfileID)
 1684    ;   User = User1
 1685    ),
 1686    Message = Dict.put(user, User).