[package]
name = "rsstg"
version = "0.1.2"
authors = ["arcade"]
edition = "2018"

[dependencies]
config = "*"

futures = "*"

create table rsstg_updates (owner integer, update jsonb);

create unique index rsstg_updates__id on rsstg_updates(update->>'update_id');

-- create table rsstg_users (id integer);
create table rsstg_channel (
	channel_id bigint primary key,
	username text);
create unique index rsstg_channel__username on rsstg_channel(username);

create table rsstg_source (
	source_id serial,
	channel_id integer not null,
	url text not null,
	last_fetch timestamptz,
	last_scrape timestamptz default now(),
	enabled boolean default false,
	iv_hash text,
	owner bigint not null);
create unique index rsstg_source__source_id on rsstg_source(source_id);
create unique index rsstg_source__channel_id__owner on rsstg_source(channel_id, owner);
create index rsstg_source__owner on rsstg_source(owner);

create table rsstg_post (
	source_id integer not null,
	date int not null,
	url text not null,
	hour smallint generated always as (extract('hour' from  posted)) stored,
	FOREIGN KEY (source_id) REFERENCES rsstg_source(source_id)
);
create unique index rsstg_post__url on rsstg_post(url);
create index rsstg_post__hour on rsstg_post(hour);
create index rsstg_post__posted_hour on rsstg_post(posted,hour);

create or replace view rsstg_order as

	}

	fn debug(&self, msg: &str) -> Result<()> {
		self.tg.spawn(SendMessage::new(self.owner_chat, msg));
		Ok(())
	}

	async fn check(&self, channel: &str, real: Option<bool>) -> Result<()> {
		match sqlx::query("select source_id, channel_id, url, last_fetch, iv_hash, owner from rsstg_source natural left join rsstg_channel where username = $1")
			.bind(channel)
			.fetch_one(&self.pool).await {
			Ok(row) => {
				let id: i32 = row.try_get("source_id")?;
				let channel_id: i64 = row.try_get("channel_id")?;
				let destination = match real {
					Some(true) => UserId::new(channel_id),
					Some(false) | None => UserId::new(row.try_get("owner")?),
				};
				let url: &str = row.try_get("url")?;
				let last_fetch: Option<DateTime<chrono::FixedOffset>> = row.try_get("last_fetch")?;
				let mut this_fetch: Option<DateTime<chrono::FixedOffset>> = None;
				let iv_hash: Option<&str> = row.try_get("iv_hash")?;
				match rss::Channel::from_url(url) {
					Ok(feed) => {
						self.debug(&format!("# title:{:?} ttl:{:?} hours:{:?} days:{:?}", feed.title(), feed.ttl(), feed.skip_hours(), feed.skip_days()))?;
						for item in feed.items() {
							let date = match item.pub_date() {
								Some(feed_date) => DateTime::parse_from_rfc2822(feed_date),
								None => DateTime::parse_from_rfc3339(&item.dublin_core_ext().unwrap().dates()[0]),
							}?;
							let url = item.link().unwrap().to_string();
							if last_fetch == None || date > last_fetch.unwrap() {
								match sqlx::query("select exists(select true from rsstg_post where url = $1 and source_id = $2) as exists;")
									.bind(&url)
									.bind(id)
									.fetch_one(&self.pool).await {
									Ok(row) => {
										let exists: bool = row.try_get("exists")?;
										if ! exists {
											if this_fetch == None || date > this_fetch.unwrap() {
												this_fetch = Some(date);
											}
											match self.tg.send( match iv_hash {
													Some(x) => SendMessage::new(destination, format!("<a href=\"https://t.me/iv?url={}&rhash={}\"> </a>{0}", url, x)),
													None => SendMessage::new(destination, format!("{}", url)),
												}.parse_mode(types::ParseMode::Html)).await {
												Ok(_) => {
													match sqlx::query("insert into rsstg_post (source_id, posted, url) values ($1, $2, $3);")
														.bind(id)
														.bind(date)
														.bind(url)
														.execute(&self.pool).await {
															Ok(_) => {},
															Err(err) => {
																self.debug(&err.to_string())?;
															},
													};
												},
												Err(err) => {
													self.debug(&err.to_string())?;
												},
											};
											tokio::time::delay_for(std::time::Duration::new(4, 0)).await;
										}
									},
									Err(err) => {
										self.debug(&err.to_string())?;
									},
								};
							};
						};
						// update last_fetch
						if this_fetch != None && (last_fetch == None || this_fetch.unwrap() > last_fetch.unwrap()) {
							match sqlx::query("update rsstg_source set last_fetch = case when (last_fetch < $1) then $1 else last_fetch end where source_id = $2;")
								.bind(this_fetch.unwrap())
								.bind(id)
								.execute(&self.pool).await {
								Ok(_) => {},
								Err(err) => {
									self.debug(&err.to_string())?;
								},
							};
						}
					},
					Err(err) => {
						self.debug(&err.to_string())?;
					},
				};
				match sqlx::query("update rsstg_source set last_scrape = now() where source_id = $1;")
					.bind(id)

				self.debug(&err.to_string())?;
			},
		};
		Ok(())
	}

	async fn clean(&self, source_id: i32) -> Result<()> {
		for query in vec!["delete from rsstg_post where source_id = $1;", "update rsstg_source set last_fetch = NULL where source_id = $1;"] {
			match sqlx::query(query)
				.bind(source_id)
				.execute(&self.pool).await {
				Ok(_) => {},
				Err(err) => {
					self.debug(&err.to_string())?;
				},
			}
		}
		Ok(())
	}

	async fn enable(&self, user: UserId, channel: &str) -> Result<()> {
		match sqlx::query("update rsstg_source set enabled = true from rsstg_channel where rsstg_channel.channel_id = rsstg_source.channel_id and rsstg_channel.username = $1 and owner = $2")
			.bind(channel)
			.bind(i64::from(user))
			.execute(&self.pool).await {
			Ok(_) => {},
			Err(err) => {
				self.debug(&err.to_string())?;
			},
		}
		Ok(())
	}

	async fn disable(&self, user: UserId, channel: &str) -> Result<()> {
		match sqlx::query("update rsstg_source set enabled = false from rsstg_channel where rsstg_channel.channel_id = rsstg_source.channel_id and rsstg_channel.username = $1 and owner = $2")
			.bind(channel)
			.bind(i64::from(user))
			.execute(&self.pool).await {
			Ok(_) => {},
			Err(err) => {
				self.debug(&err.to_string())?;
			},
		}
		Ok(())
	}

	async fn autofetch(&self) -> Result<()> {
		let mut delay = chrono::Duration::minutes(5);
		let mut next_fetch: DateTime<chrono::Local>;
		let mut now;
		loop {

			let mut rows = sqlx::query("select source_id, username, next_fetch from rsstg_order natural left join rsstg_source natural left join rsstg_channel;")
				.fetch(&self.pool);
			while let Some(row) = rows.try_next().await.unwrap() {
				now = chrono::Local::now();
				let source_id: i32 = row.try_get("source_id")?;
				next_fetch = row.try_get("next_fetch")?;
				if next_fetch < now {
					match sqlx::query("update rsstg_source set last_scrape = now() + interval '1 hour' where source_id = $1;")
						.bind(source_id)
						.execute(&self.pool).await {
						Ok(_) => {},
						Err(err) => {
							self.debug(&err.to_string())?;
						},
					};
					let clone = self.clone();
					let username: String = row.try_get("username")?;
					let username = username.clone();
					tokio::spawn(async move {
						if let Err(err) = clone.check(&username, Some(true)).await {
							eprintln!("connection error: {}", err);
						}
					});
					//&self.check(row.try_get("username")?, Some(true)).await?;
				} else {
					if next_fetch - now < delay {
						delay = next_fetch - now;
					}
				}
			};
			tokio::time::delay_for(delay.to_std()?).await;

		}
		//Ok(())
	}

}

#[tokio::main(basic_scheduler)]

						let mut words = data.split_whitespace();
						let cmd = words.next().unwrap();
						match cmd {

// start

							"/start" => {
								reply.push("Not in service yet. Try later.".to_string());
							},

// list

							"/list" => {
								reply.push("Channels:".to_string());
								let mut rows = sqlx::query("select username, enabled, url, iv_hash from rsstg_source left join rsstg_channel using (channel_id) where owner = $1")
									.bind(i64::from(message.from.id))
									.fetch(&core.pool);
								while let Some(row) = rows.try_next().await? {

									let username: &str = row.try_get("username")?;
									let enabled: bool = row.try_get("enabled")?;
									let url: &str = row.try_get("url")?;
									let iv_hash: Option<&str> = row.try_get("iv_hash")?;
									reply.push(format!("\n\\*️⃣ `{}` {}\n🔗 `{}`", username,  
										match enabled {
											true  => "🔄 enabled",
											false => "⛔ disabled",
										}, url));
									if let Some(hash) = iv_hash {
										reply.push(format!("IV `{}`", hash));
									}
								}
							},

// add

							"/add" => {




								let (channel, url, iv_hash) = (words.next().unwrap(), words.next().unwrap(), words.next());
								let ok_link = re_link.is_match(&url);
								let ok_hash = match iv_hash {
									Some(hash) => re_iv_hash.is_match(&hash),
									None => true,
								};
								if ! ok_link {

											},
											Err(err) => {
												reply.push("Sorry, unknown error\\.".to_string());
												core.debug(&format!("Sorry, unknown error: {:#?}\n", err))?;
												None
											},
									};
									match chan {
										Some(chan) => {



											match sqlx::query("insert into rsstg_source (channel_id, url, iv_hash, owner) values ($1, $2, $3, $4) on conflict (channel_id, owner) do update set url = excluded.url, iv_hash = excluded.iv_hash;")

												.bind(chan)
												.bind(url)
												.bind(iv_hash)
												.bind(i64::from(message.from.id))
												.execute(&core.pool).await {
												Ok(_) => reply.push("Channel added\\.".to_string()),
												Err(sqlx::Error::Database(err)) => {
													match err.downcast::<sqlx::postgres::PgDatabaseError>().routine() {
														Some("_bt_check_unique", ) => {
															reply.push("Duplicate key\\.".to_string());
														},
														Some(_) => {
															reply.push("Database error\\.".to_string());
														},
														None => {
															reply.push("No database error extracted\\.".to_string());
														},
													};
												},
												Err(err) => {
													reply.push("Sorry, unknown error\\.".to_string());
													core.debug(&format!("Sorry, unknown error: {:#?}\n", err))?;
												},
											};
										},
										None => {},
									};
								};
							},

// addchan

							"/addchan" => {

									};
								};
							},

// check

							"/check" => {
								let channel = words.next().unwrap();
								if ! re_username.is_match(&channel) {
									reply.push("Usernames should be something like \"@\\[a-z]\\[a-z0-9_]+\", aren't they?".to_string());
								} else {
									&core.check(channel, None).await?;
								}
							},

// clear

							"/clean" => {
								if core.owner != i64::from(message.from.id) {
									reply.push("Reserved for testing\\.".to_string());
								} else {
									let source_id = words.next().unwrap().parse::<i32>().unwrap_or(0);
									&core.clean(source_id).await?;
								}
							},

// enable

							"/enable" => {
								let channel = words.next().unwrap();
								if ! re_username.is_match(&channel) {
									reply.push("Usernames should be something like \"@\\[a-z]\\[a-z0-9_]+\", aren't they?".to_string());
								} else {
									match core.enable(message.from.id, channel).await {
										Ok(_) => {
											reply.push("Channel enabled\\.".to_string());
										}
										Err(err) => {
											core.debug(&err.to_string())?;
										},
									}
								}
							},

// disable

							"/disable" => {
								let channel = words.next().unwrap();
								if ! re_username.is_match(&channel) {
									reply.push("Usernames should be something like \"@\\[a-z]\\[a-z0-9_]+\", aren't they?".to_string());
								} else {
									match core.disable(message.from.id, channel).await {
										Ok(_) => {
											reply.push("Channel disabled\\.".to_string());
										}
										Err(err) => {
											core.debug(&err.to_string())?;
										},
									}
								}
							},

							_ => {
							},
						};
					},
					_ => {