<commit_before>// Copyright 2017-2020 <NAME>
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::config::TSConfig;
use crate::db;
use crate::error::Error;
use crate::Result;
use crate::*;
use connection::Connection;
use mysql::{Pool, PooledConn};
use std::collections::HashMap;
use std::collections::HashSet;
use std::convert::TryInto;
use std::thread::{self, sleep};
use std::time::Duration;
use std::{sync::RwLock, time::Instant};
use timer::*;
use ts3_query::*;

const CHANNEL_NAME: &str = "channel_name";
const CHANNEL_ID: &str = "cid";

const CLIENT_TYPE: &str = "client_type";
const CLIENT_TYPE_NORMAL: &str = "0";
/// Per-Identity ID
const CLIENT_DB_ID: &str = "client_database_id";
/// Per-connection ID
const CLIENT_CONN_ID: &str = "clid";
const CLIENT_GROUPS: &str = "client_servergroups";
const CLIENT_CHANNEL: &str = CHANNEL_ID;
const CLIENT_NAME: &str = "client_nickname";

/// Safety: connection timeout has to be short enough that no request blocks others!
const INTERVAL_ACTIVITY_S: i64 = 30;
const INTERVAL_FLUSH_M: i64 = 15;
const COOLDOWN_POKE_S: u64 = 60 * 20;

mod connection;

/// Holds TS statistics data and cleans up on drop
///
/// Is used by daemon to hold user activity.
struct TsStatCtrl {
    pool: Pool,
    conn: Connection,
    names: HashMap<TsClDBID, String>,
    times: HashMap<(TsClDBID, TsChannelID), i32>,
    last_channel: HashMap<TsClDBID, TsChannelID>,
    last_date: chrono::naive::NaiveDate,
    last_update: Instant,
    last_guest_poke: Option<Instant>,
    /// none also if no poke enabled
    poke_config: Option<PokeConfig>,
}

#[derive(Default, Debug)]
struct PokeConfig {
    poke_group: TsGroupID,
    guest_group: TsGroupID,
    guest_channel: TsGroupID,
    poke_msg: String,
}

impl TsStatCtrl {
    fn new(pool: Pool, config: Config) -> Result<Self> {
        let mut data = Self {
            pool,
            conn: Connection::new(config)?,
            last_date: Local::today().naive_local(),
            last_update: Instant::now(),
            names: Default::default(),
            times: Default::default(),
            last_channel: Default::default(),
            last_guest_poke: Default::default(),
            poke_config: Default::default(),
        };
        data.inner_update_settings(true)?;
        Ok(data)
    }

    /// Update dynamic settings from DB
    pub fn update_settings(&mut self) -> Result<()> {
        self.inner_update_settings(false)
    }

    /// Update dynamic settings from DB. first_run controls log-output
    fn inner_update_settings(&mut self, first_run: bool) -> Result<()> {
        // shouldn't name Pool & TsConnection conn..
        let old_enabled = self.poke_config.is_some();
        let mut conn = self.pool.get_conn()?;
        self.poke_config = if is_ts3_guest_check_enabled(&mut conn, &self.conn.config()) {
            read_poke_config(&mut conn)?
        } else {
            None
        };
        if self.poke_config.is_some() != old_enabled || first_run {
            let msg = if self.poke_config.is_some() {
                "Ts-Poke enabled"
            } else {
                "Ts-Poke disabled"
            };
            db::log_message(&mut conn, msg, "unable to log poke config change");
        }
        Ok(())
    }

    /// Check online clients, update activity & send notifications
    fn tick(&mut self) -> Result<()> {
        // store timestamp now to prevent delta loss by blocking operations
        let new_timestamp = Instant::now();
        let data = get_online_clients(&mut self.conn)?;
        // take elapsed after data, expecting server reponse to be fast and connection start possibly slow
        let elapsed: i32 = match self.last_update.elapsed().as_secs().try_into() {
            Ok(v) => v,
            Err(e) => panic!("TS activity elapsed time > i32::max! {}", e),
        };
        trace!("Elapsed: {} seconds", elapsed);
        self.last_update = new_timestamp;

        let mut new_channel = HashMap::with_capacity(data.len());

        let mut poke_clients: Vec<TsConID> = Vec::new();
        let mut poke = false;
        let poke_cfg = if self
            .last_guest_poke
            .map_or(false, |v| v.elapsed().as_secs() > COOLDOWN_POKE_S)
        {
            self.poke_config.as_ref()
        } else {
            None
        };

        for client in data {
            let id = client.cldbid;
            self.names.insert(id, client.name);
            // add elapsed time to last channel, or current if no previous is known
            let chan = self.last_channel.get(&id).unwrap_or(&client.channel);

            let k = (id, *chan);
            if let Some(time) = self.times.get_mut(&k) {
                *time += elapsed;
            } else {
                self.times.insert(k, elapsed);
            }

            if let Some(ref cfg) = poke_cfg {
                if cfg.guest_channel == client.channel && client.groups.contains(&cfg.guest_group) {
                    poke = true;
                } else if client.groups.contains(&cfg.poke_group) {
                    poke_clients.push(client.conid);
                }
            }

            // remember current channel for next time
            new_channel.insert(id, client.channel);
            // TODO: group check
        }

        // excludes disconnected clients
        self.last_channel = new_channel;

        if poke {
            self.poke_clients(poke_clients)?;
        }
        Ok(())
    }

    /// poke clients, runs new thread
    fn poke_clients(&mut self, clients: Vec<TsConID>) -> Result<()> {
        self.last_guest_poke = Some(Instant::now());
        let cooldown = match self.conn.config().ts.cmd_limit_secs {
            0 => None,
            // subtract
            v => Some(Duration::from_millis(1000 / (v as u64))),
        };
        let msg: &str = match self.poke_config.as_ref().map(|v| &v.poke_msg) {
            Some(v) => v,
            None => unreachable!("Unreachable! poke_clients invoked without poke_config!"),
        };
        let escaped = raw::escape_arg(msg);
        let mut conn = self.conn.clone()?;
        thread::Builder::new()
            .name("ts-poke".to_string())
            .spawn(move || {
                let res = || -> Result<()> {
                    for client in clients {
                        if let Some(cooldown) = cooldown {
                            sleep(cooldown);
                        }
                        conn.get()?
                            .raw_command(&format!("poke clid={} msg={}", client, &escaped))?;
                    }
                    Ok(())<commit_msg>Check for lt in guest-channel before poking<commit_after>// Copyright 2017-2020 <NAME>
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::config::TSConfig;
use crate::db;
use crate::error::Error;
use crate::Result;
use crate::*;
use connection::Connection;
use mysql::{Pool, PooledConn};
use std::collections::HashMap;
use std::collections::HashSet;
use std::convert::TryInto;
use std::thread::{self, sleep};
use std::time::Duration;
use std::{sync::RwLock, time::Instant};
use timer::*;
use ts3_query::*;

const CHANNEL_NAME: &str = "channel_name";
const CHANNEL_ID: &str = "cid";

const CLIENT_TYPE: &str = "client_type";
const CLIENT_TYPE_NORMAL: &str = "0";
/// Per-Identity ID
const CLIENT_DB_ID: &str = "client_database_id";
/// Per-connection ID
const CLIENT_CONN_ID: &str = "clid";
const CLIENT_GROUPS: &str = "client_servergroups";
const CLIENT_CHANNEL: &str = CHANNEL_ID;
const CLIENT_NAME: &str = "client_nickname";

/// Safety: connection timeout has to be short enough that no request blocks others!
const INTERVAL_ACTIVITY_S: i64 = 30;
const INTERVAL_FLUSH_M: i64 = 15;
const COOLDOWN_POKE_S: u64 = 60 * 20;

mod connection;

/// Holds TS statistics data and cleans up on drop
///
/// Is used by daemon to hold user activity.
struct TsStatCtrl {
    pool: Pool,
    conn: Connection,
    names: HashMap<TsClDBID, String>,
    times: HashMap<(TsClDBID, TsChannelID), i32>,
    last_channel: HashMap<TsClDBID, TsChannelID>,
    last_date: chrono::naive::NaiveDate,
    last_update: Instant,
    last_guest_poke: Option<Instant>,
    /// none also if no poke enabled
    poke_config: Option<PokeConfig>,
}

#[derive(Default, Debug)]
struct PokeConfig {
    poke_group: TsGroupID,
    guest_group: TsGroupID,
    guest_channel: TsGroupID,
    poke_msg: String,
}

impl TsStatCtrl {
    fn new(pool: Pool, config: Config) -> Result<Self> {
        let mut data = Self {
            pool,
            conn: Connection::new(config)?,
            last_date: Local::today().naive_local(),
            last_update: Instant::now(),
            names: Default::default(),
            times: Default::default(),
            last_channel: Default::default(),
            last_guest_poke: Default::default(),
            poke_config: Default::default(),
        };
        data.inner_update_settings(true)?;
        Ok(data)
    }

    /// Update dynamic settings from DB
    pub fn update_settings(&mut self) -> Result<()> {
        self.inner_update_settings(false)
    }

    /// Update dynamic settings from DB. first_run controls log-output
    fn inner_update_settings(&mut self, first_run: bool) -> Result<()> {
        // shouldn't name Pool & TsConnection conn..
        let old_enabled = self.poke_config.is_some();
        let mut conn = self.pool.get_conn()?;
        self.poke_config = if is_ts3_guest_check_enabled(&mut conn, &self.conn.config()) {
            read_poke_config(&mut conn)?
        } else {
            None
        };
        if self.poke_config.is_some() != old_enabled || first_run {
            let msg = if self.poke_config.is_some() {
                "Ts-Poke enabled"
            } else {
                "Ts-Poke disabled"
            };
            db::log_message(&mut conn, msg, "unable to log poke config change");
        }
        Ok(())
    }

    /// Check online clients, update activity & send notifications
    fn tick(&mut self) -> Result<()> {
        // store timestamp now to prevent delta loss by blocking operations
        let new_timestamp = Instant::now();
        let data = get_online_clients(&mut self.conn)?;
        // take elapsed after data, expecting server reponse to be fast and connection start possibly slow
        let elapsed: i32 = match self.last_update.elapsed().as_secs().try_into() {
            Ok(v) => v,
            Err(e) => panic!("TS activity elapsed time > i32::max! {}", e),
        };
        trace!("Elapsed: {} seconds", elapsed);
        self.last_update = new_timestamp;

        let mut new_channel = HashMap::with_capacity(data.len());

        let mut poke_clients: Vec<TsConID> = Vec::new();
        // new guest
        let mut new_guest = false;
        // poke-receiver at place
        let mut has_lt = false;
        let poke_cfg = if self
            .last_guest_poke
            .map_or(false, |v| v.elapsed().as_secs() > COOLDOWN_POKE_S)
        {
            self.poke_config.as_ref()
        } else {
            None
        };

        for client in data {
            let id = client.cldbid;
            self.names.insert(id, client.name);
            // add elapsed time to last channel, or current if no previous is known
            let chan = self.last_channel.get(&id).unwrap_or(&client.channel);

            let k = (id, *chan);
            if let Some(time) = self.times.get_mut(&k) {
                *time += elapsed;
            } else {
                self.times.insert(k, elapsed);
            }

            if let Some(cfg) = poke_cfg {
                if !has_lt {
                    if cfg.guest_channel == client.channel {
                        if client.groups.contains(&cfg.guest_group) {
                            new_guest = true;
                        } else if client.groups.contains(&cfg.guest_group) {
                            has_lt = true;
                        }
                    } else if client.groups.contains(&cfg.poke_group) {
                        poke_clients.push(client.conid);
                    }
                }
            }

            // remember current channel for next time
            new_channel.insert(id, client.channel);
        }

        // excludes disconnected clients
        self.last_channel = new_channel;

        if new_guest && !has_lt {
            self.poke_clients(poke_clients)?;
        } else if has_lt {
            // reset cooldown time if poke-receiver at place
            self.last_guest_poke = Some(Instant::now());
        }
        Ok(())
    }

    /// poke clients, runs new thread
    fn poke_clients(&mut self, clients: Vec<TsConID>) -> Result<()> {
        self.last_guest_poke = Some(Instant::now());
        let cooldown = match self.conn.config().ts.cmd_limit_secs {
            0 => None,
            // subtract
            v => Some(Duration::from_millis(1000 / (v as u64))),
        };
        let msg: &str = match self.poke_config.as_ref().map(|v| &v.poke_msg) {
            Some(v) => v,
            None => unreachable!("Unreachable! poke_clients invoked without poke_config!"),
        };
        let escaped = raw::escape_arg(msg);
        let mut conn = self.conn.clone()?;
        thread::Builder::new()
            .name("ts-poke".to_string())
            .spawn(move || {
                let res = || -> Result<()> {
                    for client in clients {
                        if let Some(cooldown) = cooldown {
                            sleep(cooldown);
                        }
                        conn.get()?
                            .raw_command(&format!("poke clid={} msg={}", client, &escaped))?;
                    }
                    Ok(())
{'fraction_non_alphanumeric': 0.12141984525587078, 'fraction_numerical': 0.005701099497760282, 'mean_word_length': 2.1132474117895628, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '1263566', 'n_tokens_mistral': 4473, 'n_tokens_neox': 4325, 'n_words': 1171}