diff --git a/.github/workflows/lint-migrations.yml b/.github/workflows/lint-migrations.yml index c1587f696..00350abcd 100644 --- a/.github/workflows/lint-migrations.yml +++ b/.github/workflows/lint-migrations.yml @@ -10,11 +10,13 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v1 + - name: Fetch main branch + run: git fetch origin main:main - name: Find modified migrations run: | - modified_migrations=$(git diff --diff-filter=d --name-only origin/$GITHUB_BASE_REF...origin/$GITHUB_HEAD_REF 'packages/db/migrations/*.do.*.sql') + modified_migrations=$(git diff --diff-filter=d --name-only main 'packages/db/migrations/*.do.*.sql') echo "$modified_migrations" - echo "::set-output name=file_names::$modified_migrations" + echo "file_names=$modified_migrations" >> $GITHUB_OUTPUT id: modified-migrations - uses: sbdchd/squawk-action@v1 with: diff --git a/ml/digest-score/Dockerfile b/ml/digest-score/Dockerfile new file mode 100644 index 000000000..0afa6baec --- /dev/null +++ b/ml/digest-score/Dockerfile @@ -0,0 +1,13 @@ +FROM python:3.8-slim + +WORKDIR /app + +ENV GRPC_PYTHON_BUILD_SYSTEM_OPENSSL "1" +ENV GRPC_PYTHON_BUILD_SYSTEM_ZLIB "1" + +COPY . /app + +RUN pip install --no-cache-dir -r requirements.txt + +EXPOSE 5000 +CMD ["python", "serve.py"] diff --git a/ml/digest-score/app.py b/ml/digest-score/app.py index 0d1765525..f9c428100 100644 --- a/ml/digest-score/app.py +++ b/ml/digest-score/app.py @@ -1,8 +1,5 @@ -import psycopg2 - import logging from flask import Flask, request, jsonify -from pydantic import BaseModel, ConfigDict, ValidationError, conlist from typing import List @@ -10,58 +7,23 @@ import os import sys import json import pytz +import pickle import numpy as np import pandas as pd import joblib -from datetime import datetime, timedelta +from urllib.parse import urlparse from datetime import datetime import dateutil.parser from google.cloud import storage +from features.user_history import FEATURE_COLUMNS app = Flask(__name__) logging.basicConfig(level=logging.INFO, stream=sys.stdout) -TRAIN_FEATURES = [ - "item_has_thumbnail", - "item_has_site_icon", +USER_HISTORY_PATH = 'user_features.pkl' +MODEL_PIPELINE_PATH = 'predict_read_pipeline-v002.pkl' - 'user_30d_interactions_author_count', - 'user_30d_interactions_site_count', - 'user_30d_interactions_subscription_count', - - 'user_30d_interactions_author_rate', - 'user_30d_interactions_site_rate', - 'user_30d_interactions_subscription_rate', - - 'global_30d_interactions_site_count', - 'global_30d_interactions_author_count', - 'global_30d_interactions_subscription_count', - - 'global_30d_interactions_site_rate', - 'global_30d_interactions_author_rate', - 'global_30d_interactions_subscription_rate' -] - -DB_PARAMS = { - 'dbname': os.getenv('DB_NAME') or 'omnivore', - 'user': os.getenv('DB_USER'), - 'password': os.getenv('DB_PASSWORD'), - 'host': os.getenv('DB_HOST') or 'localhost', - 'port': os.getenv('DB_PORT') or '5432' -} - -USER_FEATURES = { - "site": "user_30d_interactions_site", - "author": "user_30d_interactions_author", - "subscription": "user_30d_interactions_subscription", -} - -GLOBAL_FEATURES = { - "site": "global_30d_interactions_site", - "author": "global_30d_interactions_author", - "subscription": "global_30d_interactions_subscription", -} def download_from_gcs(bucket_name, gcs_path, destination_path): storage_client = storage.Client() @@ -70,94 +32,44 @@ def download_from_gcs(bucket_name, gcs_path, destination_path): blob.download_to_filename(destination_path) -def load_pipeline(): - bucket_name = os.getenv('GCS_BUCKET') - pipeline_gcs_path = os.getenv('PIPELINE_GCS_PATH') - download_from_gcs(bucket_name, pipeline_gcs_path, '/tmp/pipeline.pkl') - pipeline = joblib.load('/tmp/pipeline.pkl') +def load_pipeline(path): + pipeline = joblib.load(path) return pipeline -def load_pipeline_local(): - pipeline = joblib.load('predict_user_clicked_random_forest_pipeline-v001.pkl') - return pipeline +def load_tables_from_pickle(path): + with open(path, 'rb') as handle: + tables = pickle.load(handle) + return tables -def fetch_user_features(name, feature_name): - conn = psycopg2.connect(**DB_PARAMS) - cur = conn.cursor() - query = f"SELECT user_id, {name}, interactions, interaction_rate FROM {feature_name}" - - cur.execute(query) - data = cur.fetchall() - - cur.close() - conn.close() - columns = [ - "user_id", - name, - "interactions", - "interaction_rate" - ] - - rate_feature_name = f"{feature_name}_rate" - count_feature_name = f"{feature_name}_count" - - df_loaded = pd.DataFrame(data, columns=columns) - df_loaded = df_loaded.rename(columns={"interactions": count_feature_name}, errors="raise") - df_loaded = df_loaded.rename(columns={"interaction_rate": rate_feature_name}, errors="raise") - df_loaded[rate_feature_name] = df_loaded[rate_feature_name].fillna(0) - df_loaded[count_feature_name] = df_loaded[count_feature_name].fillna(0) - - return df_loaded - - -def fetch_global_features(name, feature_name): - conn = psycopg2.connect(**DB_PARAMS) - cur = conn.cursor() - query = f"SELECT {name}, interactions, interaction_rate FROM {feature_name}" - - cur.execute(query) - data = cur.fetchall() - - cur.close() - conn.close() - columns = [ - name, - "interactions", - "interaction_rate" - ] - - rate_feature_name = f"{feature_name}_rate" - count_feature_name = f"{feature_name}_count" - - df_loaded = pd.DataFrame(data, columns=columns) - df_loaded = df_loaded.rename(columns={"interactions": count_feature_name}, errors="raise") - df_loaded = df_loaded.rename(columns={"interaction_rate": rate_feature_name}, errors="raise") - df_loaded[rate_feature_name] = df_loaded[rate_feature_name].fillna(0) - df_loaded[count_feature_name] = df_loaded[count_feature_name].fillna(0) - - return df_loaded - - -def load_user_features(): +def load_user_features(path): result = {} - for view_name in USER_FEATURES.keys(): - key_name = USER_FEATURES[view_name] - result[key_name] = fetch_user_features(view_name, key_name) - app.logger.info(f"loaded {len(result[key_name])} features for {key_name}") + tables = load_tables_from_pickle(path) + for table_name in tables.keys(): + result[table_name] = tables[table_name].to_pandas() return result -def load_global_features(): +def dataframe_to_dict(df): result = {} - for view_name in GLOBAL_FEATURES.keys(): - key_name = GLOBAL_FEATURES[view_name] - result[key_name] = fetch_global_features(view_name, key_name) - app.logger.info(f"loaded {len(result[key_name])} features for {key_name}") + for index, row in df.iterrows(): + user_id = row['user_id'] + if user_id not in result: + result[user_id] = [] + result[user_id].append(row.to_dict()) return result +def merge_dicts(dict1, dict2): + for key, value in dict2.items(): + if key in dict1: + dict1[key].extend(value) + else: + dict1[key] = value + return dict1 + + def compute_score(user_id, item_features): interaction_score = compute_interaction_score(user_id, item_features) return { @@ -166,86 +78,55 @@ def compute_score(user_id, item_features): } -def compute_time_bonus_score(item_features): - saved_at = item_features['saved_at'] - current_time = datetime.now(pytz.utc) - time_diff_hours = (current_time - saved_at).total_seconds() / 3600 - max_diff_hours = 3 * 24 - if time_diff_hours >= max_diff_hours: - return 0.0 - else: - return max(0.0, min(1.0, 1 - (time_diff_hours / max_diff_hours))) - - def compute_interaction_score(user_id, item_features): - print('item_features', item_features) + original_url_host = urlparse(item_features.get('original_url')).netloc df_test = pd.DataFrame([{ 'user_id': user_id, 'author': item_features.get('author'), 'site': item_features.get('site'), 'subscription': item_features.get('subscription'), + 'original_url_host': original_url_host, 'item_has_thumbnail': 1 if item_features.get('has_thumbnail') else 0, "item_has_site_icon": 1 if item_features.get('has_site_icon') else 0, + + 'item_word_count': item_features.get('words_count'), + 'is_subscription': 1 if item_features.get('is_subscription') else 0, + 'is_newsletter': 1 if item_features.get('is_newsletter') else 0, + 'is_feed': 1 if item_features.get('is_feed') else 0, + 'days_since_subscribed': item_features.get('days_since_subscribed'), + 'subscription_count': item_features.get('subscription_count'), + 'subscription_auto_add_to_library': item_features.get('subscription_auto_add_to_library'), + 'subscription_fetch_content': item_features.get('subscription_fetch_content'), + + 'has_author': 1 if item_features.get('author') else 0, + 'inbox_folder': 1 if item_features.get('folder') == 'inbox' else 0, }]) - for name in USER_FEATURES.keys(): - feature_name = USER_FEATURES[name] - df_feature = user_features[feature_name] - df_test = df_test.merge(df_feature, on=['user_id', name], how='left') - df_test[f"{feature_name}_rate"] = df_test[f"{feature_name}_rate"].fillna(0) - df_test[f"{feature_name}_count"] = df_test[f"{feature_name}_count"].fillna(0) + for name, df in user_features.items(): + df = df[df['user_id'] == user_id] + if 'author' in name: + merge_keys = ['user_id', 'author'] + elif 'site' in name: + merge_keys = ['user_id', 'site'] + elif 'subscription' in name: + merge_keys = ['user_id', 'subscription'] + elif 'original_url_host' in name: + merge_keys = ['user_id', 'original_url_host'] + else: + print("skipping feature: ", name) + continue - for name in GLOBAL_FEATURES.keys(): - feature_name = GLOBAL_FEATURES[name] - df_feature = global_features[feature_name] - df_test = df_test.merge(df_feature, on=name, how='left') - df_test[f"{feature_name}_rate"] = df_test[f"{feature_name}_rate"].fillna(0) - df_test[f"{feature_name}_count"] = df_test[f"{feature_name}_count"].fillna(0) + df_test = pd.merge(df_test, df, on=merge_keys, how='left') + df_test = df_test.fillna(0) + df_predict = df_test[FEATURE_COLUMNS] - df_predict = df_test[TRAIN_FEATURES] - - # Print out the columns with values, so we can know how sparse our data is - #scored_columns = df_predict.columns[(df_predict.notnull() & (df_predict != 0)).any()].tolist() - #print("scored columns", scored_columns) interaction_score = pipeline.predict_proba(df_predict) + print('score', interaction_score, 'item_features', df_test[df_test != 0].stack()) return interaction_score[0][1] -def get_library_item(library_item_id): - conn = psycopg2.connect(**DB_PARAMS) - cur = conn.cursor() - query = """ - SELECT - li.title, - li.author, - li.saved_at, - li.site_name as site, - li.item_language as language, - li.subscription, - li.word_count, - li.directionality, - CASE WHEN li.thumbnail IS NOT NULL then 1 else 0 END as has_thumbnail, - CASE WHEN li.site_icon IS NOT NULL then 1 else 0 END as has_site_icon - FROM omnivore.library_item li - WHERE li.id = %s - """ - - cur.execute(query, (library_item_id,)) - - data = cur.fetchone() - columns = [desc[0] for desc in cur.description] - cur.close() - conn.close() - - if data: - item_dict = dict(zip(columns, data)) - return item_dict - else: - return None - - @app.route('/_ah/health', methods=['GET']) def ready(): return jsonify({'OK': 'yes'}), 200 @@ -254,29 +135,17 @@ def ready(): @app.route('/users//features', methods=['GET']) def get_user_features(user_id): result = {} + df_user = pd.DataFrame([{ + 'user_id': user_id, + }]) - for name in USER_FEATURES.keys(): - feature_name = USER_FEATURES[name] - rate_feature_name = f"{feature_name}_rate" - count_feature_name = f"{feature_name}_count" - df_feature = user_features[feature_name] - df_filtered = df_feature[df_feature['user_id'] == user_id] - if not df_filtered.empty: - rate = df_filtered[[name, rate_feature_name]].dropna().to_dict(orient='records') - count = df_filtered[[name, count_feature_name]].dropna().to_dict(orient='records') - result[feature_name] = { - 'rate': rate, - 'count': count - } + user_data = {} + for name, df in user_features.items(): + df = df[df['user_id'] == user_id] + df_dict = dataframe_to_dict(df) + user_data = merge_dicts(user_data, df_dict) - return jsonify(result), 200 - - -@app.route('/users//library_items//score', methods=['GET']) -def get_library_item_score(user_id, library_item_id): - item_features = get_library_item(library_item_id) - score = compute_score(user_id, item_features) - return jsonify({'score': score}) + return jsonify(user_data), 200 @app.route('/predict', methods=['POST']) @@ -287,7 +156,6 @@ def predict(): user_id = data.get('user_id') item_features = data.get('item_features') - item_features['saved_at'] = dateutil.parser.isoparse(item_features['saved_at']) if user_id is None: return jsonify({'error': 'Missing user_id'}), 400 @@ -316,7 +184,6 @@ def batch(): print('key": ', key) print('item: ', item) library_item_id = item['library_item_id'] - item['saved_at'] = dateutil.parser.isoparse(item['saved_at']) result[library_item_id] = compute_score(user_id, item) return jsonify(result) @@ -325,14 +192,15 @@ def batch(): return jsonify({'error': str(e)}), 500 -if os.getenv('LOAD_LOCAL_MODEL'): - pipeline = load_pipeline_local() -else: - pipeline = load_pipeline() +if os.getenv('LOAD_LOCAL_MODEL') != None: + gcs_bucket_name = os.getenv('GCS_BUCKET') + download_from_gcs(gcs_bucket_name, f'data/features/user_features.pkl', USER_HISTORY_PATH) + download_from_gcs(gcs_bucket_name, f'data/models/predict_read_pipeline-v002.pkl', MODEL_PIPELINE_PATH) -user_features = load_user_features() -global_features = load_global_features() +pipeline = load_pipeline(MODEL_PIPELINE_PATH) +user_features = load_user_features(USER_HISTORY_PATH) +print('loaded pipeline and user_features', pipeline, user_features) if __name__ == '__main__': app.run(debug=True, port=5000) \ No newline at end of file diff --git a/ml/digest-score/features.py b/ml/digest-score/features.py new file mode 100644 index 000000000..47766c244 --- /dev/null +++ b/ml/digest-score/features.py @@ -0,0 +1,31 @@ +import psycopg2 +import numpy as np +import pandas as pd +from sqlalchemy import create_engine, text +from datetime import datetime, timedelta + +import os +from io import BytesIO +import tempfile + +import pyarrow as pa +import pyarrow.parquet as pq +from google.cloud import storage + +from features.extract import extract_and_upload_raw_data +from features.user_history import generate_and_upload_user_history + + + +def main(): + execution_date = os.getenv('EXECUTION_DATE') + num_days_history = os.getenv('NUM_DAYS_HISTORY') + gcs_bucket_name = os.getenv('GCS_BUCKET') + + extract_and_upload_raw_data(execution_date, num_days_history, gcs_bucket_name) + generate_and_upload_user_history(execution_date, gcs_bucket_name) + + print("done") + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/ml/digest-score/features/__init__.py b/ml/digest-score/features/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/ml/digest-score/features/extract.py b/ml/digest-score/features/extract.py new file mode 100644 index 000000000..58e70ecf8 --- /dev/null +++ b/ml/digest-score/features/extract.py @@ -0,0 +1,109 @@ +# extract and upload raw data used for feature generation + +import psycopg2 +import numpy as np +import pandas as pd +from sqlalchemy import create_engine, text +from datetime import datetime, timedelta + +import os +from io import BytesIO +import tempfile + +import pyarrow as pa +import pyarrow.parquet as pq +from google.cloud import storage + +DB_PARAMS = { + 'dbname': os.getenv('DB_NAME') or 'omnivore', + 'user': os.getenv('DB_USER'), + 'password': os.getenv('DB_PASSWORD'), + 'host': os.getenv('DB_HOST') or 'localhost', + 'port': os.getenv('DB_PORT') or '5432' +} + +def extract_host(url): + try: + return urlparse(url).netloc + except Exception as e: + return None + +def fetch_raw_data(date_str, num_days_history): + end_date = pd.to_datetime(date_str) + start_date = end_date - timedelta(days=num_days_history) + start_date_str = start_date.strftime('%Y-%m-%d 00:00:00') + end_date_str = end_date.strftime('%Y-%m-%d 23:59:59') + + conn_str = f"postgresql://{DB_PARAMS['user']}:{DB_PARAMS['password']}@{DB_PARAMS['host']}:{DB_PARAMS['port']}/{DB_PARAMS['dbname']}" + # conn_str = f"postgresql://{DB_PARAMS['host']}:{DB_PARAMS['port']}/{DB_PARAMS['dbname']}" + engine = create_engine(conn_str) + + query = text(""" + SELECT + li.id as library_item_id, + li.user_id, + li.created_at, + li.archived_at, + li.deleted_at, + CASE WHEN li.folder = 'inbox' then 1 else 0 END as inbox_folder, + li.item_type, + li.item_language AS language, + li.content_reader, + li.word_count as item_word_count, + CASE WHEN li.thumbnail IS NOT NULL then 1 else 0 END as item_has_thumbnail, + CASE WHEN li.site_icon IS NOT NULL then 1 else 0 END as item_has_site_icon, + li.original_url, + li.site_name AS site, + li.author, + li.subscription, + sub.type as subscription_type, + sub.created_at as subscription_start_date, + sub.count as subscription_count, + sub.auto_add_to_library as subscription_auto_add_to_library, + sub.fetch_content as subscription_fetch_content, + sub.folder as subscription_folder, + CASE WHEN li.read_at is not NULL then 1 else 0 END as user_clicked, + CASE WHEN li.reading_progress_bottom_percent > 10 THEN 1 ELSE 0 END AS user_read, + CASE WHEN li.reading_progress_bottom_percent > 50 THEN 1 ELSE 0 END AS user_long_read + FROM omnivore.library_item AS li + LEFT JOIN omnivore.subscriptions sub on li.subscription = sub.name AND sub.user_id = li.user_id + WHERE li.created_at >= :start_date AND li.created_at <= :end_date; + """) + + chunk_size = 100000 # Adjust based on available memory and performance needs + + with tempfile.TemporaryDirectory() as tmpdir: + parquet_files = [] + with engine.connect() as conn: + for i, chunk in enumerate(pd.read_sql(query, conn, params={'start_date': start_date_str, 'end_date': end_date_str}, chunksize=chunk_size)): + chunk['library_item_id'] = chunk['library_item_id'].astype(str) + chunk['user_id'] = chunk['user_id'].astype(str) + chunk['original_url_host'] = chunk['original_url'].apply(extract_host) + + parquet_file = os.path.join(tmpdir, f'chunk_{i}.parquet') + chunk.to_parquet(parquet_file) + parquet_files.append(parquet_file) + + concatenated_df = pd.concat([pd.read_parquet(file) for file in parquet_files], ignore_index=True) + + parquet_buffer = BytesIO() + table = pa.Table.from_pandas(concatenated_df) + pq.write_table(table, parquet_buffer) + parquet_buffer.seek(0) + + return parquet_buffer + + +def upload_raw_databuffer(feather_buffer, execution_date, gcs_bucket_name): + client = storage.Client() + bucket = client.bucket(gcs_bucket_name) + blob = bucket.blob(f'data/raw/library_items_{execution_date}.parquet') + blob.upload_from_file(feather_buffer, content_type='application/octet-stream') + + print("Data stored successfully.") + + +def extract_and_upload_raw_data(execution_date, num_days_history, gcs_bucket_name): + buffer = fetch_raw_data(execution_date, int(num_days_history)) + upload_raw_databuffer(buffer, execution_date, gcs_bucket_name) + diff --git a/ml/digest-score/features/user_history.py b/ml/digest-score/features/user_history.py new file mode 100644 index 000000000..289ecc64d --- /dev/null +++ b/ml/digest-score/features/user_history.py @@ -0,0 +1,222 @@ +# download raw user data, aggregate user history, and upload to GCS + +import psycopg2 +import numpy as np +import pandas as pd +from sqlalchemy import create_engine, text +from datetime import datetime, timedelta + +import os +from io import BytesIO +import tempfile + +import pickle +import pyarrow as pa +import pyarrow.parquet as pq +import pyarrow.feather as feather +from google.cloud import storage + +FEATURE_COLUMNS=[ + # targets + # 'user_clicked', 'user_read', 'user_long_read', + + # item attributes / user setup attributes + 'item_word_count','item_has_site_icon', 'is_subscription', + 'inbox_folder', 'has_author', + + # how the user has setup the subscription + 'is_newsletter', 'is_feed', 'days_since_subscribed', + 'subscription_count', 'subscription_auto_add_to_library', + 'subscription_fetch_content', + + # user/item interaction history + 'user_original_url_host_saved_count_week_1', + 'user_original_url_host_interaction_count_week_1', + 'user_original_url_host_rate_week_1', + 'user_original_url_host_proportion_week_1', + + 'user_original_url_host_saved_count_week_2', + 'user_original_url_host_interaction_count_week_2', + 'user_original_url_host_rate_week_2', + 'user_original_url_host_proportion_week_2', + 'user_original_url_host_saved_count_week_3', + 'user_original_url_host_interaction_count_week_3', + 'user_original_url_host_rate_week_3', + 'user_original_url_host_proportion_week_3', + 'user_original_url_host_saved_count_week_4', + 'user_original_url_host_interaction_count_week_4', + 'user_original_url_host_rate_week_4', + 'user_original_url_host_proportion_week_4', + + 'user_subscription_saved_count_week_1', + 'user_subscription_interaction_count_week_1', + 'user_subscription_rate_week_1', 'user_subscription_proportion_week_1', + 'user_site_saved_count_week_3', 'user_site_interaction_count_week_3', + 'user_site_rate_week_3', 'user_site_proportion_week_3', + 'user_site_saved_count_week_2', 'user_site_interaction_count_week_2', + 'user_site_rate_week_2', 'user_site_proportion_week_2', + 'user_subscription_saved_count_week_2', + 'user_subscription_interaction_count_week_2', + 'user_subscription_rate_week_2', 'user_subscription_proportion_week_2', + 'user_site_saved_count_week_1', 'user_site_interaction_count_week_1', + 'user_site_rate_week_1', 'user_site_proportion_week_1', + 'user_subscription_saved_count_week_3', + 'user_subscription_interaction_count_week_3', + 'user_subscription_rate_week_3', 'user_subscription_proportion_week_3', + 'user_author_saved_count_week_4', + 'user_author_interaction_count_week_4', 'user_author_rate_week_4', + 'user_author_proportion_week_4', 'user_author_saved_count_week_1', + 'user_author_interaction_count_week_1', 'user_author_rate_week_1', + 'user_author_proportion_week_1', 'user_site_saved_count_week_4', + 'user_site_interaction_count_week_4', 'user_site_rate_week_4', + 'user_site_proportion_week_4', 'user_author_saved_count_week_2', + 'user_author_interaction_count_week_2', 'user_author_rate_week_2', + 'user_author_proportion_week_2', 'user_author_saved_count_week_3', + 'user_author_interaction_count_week_3', 'user_author_rate_week_3', + 'user_author_proportion_week_3', 'user_subscription_saved_count_week_4', + 'user_subscription_interaction_count_week_4', + 'user_subscription_rate_week_4', 'user_subscription_proportion_week_4' +] + +def parquet_to_dataframe(file_path): + table = pq.read_table(file_path) + df = table.to_pandas() + return df + + +def load_tables_from_pickle(pickle_file): + with open(pickle_file, 'rb') as handle: + tables = pickle.load(handle) + return tables + + +def download_raw_library_items(execution_date, gcs_bucket_name): + local_file_path = 'raw_library_items.parquet' + + client = storage.Client() + bucket = client.bucket(gcs_bucket_name) + blob = bucket.blob(f'data/raw/library_items_{execution_date}.parquet') + blob.download_to_filename(local_file_path) + + df = parquet_to_dataframe(local_file_path) + + os.remove(local_file_path) + return df + + +def load_feather_files(feature_directory): + dataframes = {} + for file_name in os.listdir(feature_directory): + if file_name.endswith('.feather'): + file_path = os.path.join(feature_directory, file_name) + df_name = os.path.splitext(file_name)[0] # Use the file name (without extension) as key + table = feather.read_table(file_path) + dataframes[df_name] = table + return dataframes + + +def save_tables_to_arrow_ipc_with_schemas(tables, output_file): + with pa.OSFile(output_file, 'wb') as sink: + with pa.ipc.new_stream(sink, pa.schema([])) as writer: + for name, table in tables.items(): + metadata = table.schema.metadata or {} + metadata = {**metadata, b'table_name': name.encode('utf-8')} + schema = table.schema.add_metadata(metadata) + print("NAME:", name, "TABLE", table) + writer.write_table(table.replace_schema_metadata(schema.metadata)) + + +def save_tables_to_pickle(tables, output_file): + with open(output_file, 'wb') as handle: + pickle.dump(tables, handle, protocol=pickle.HIGHEST_PROTOCOL) + + +def upload_to_gcs(bucket_name, source_file_name, destination_blob_name): + client = storage.Client() + bucket = client.bucket(bucket_name) + blob = bucket.blob(destination_blob_name) + blob.upload_from_filename(source_file_name) + print(f'File {source_file_name} uploaded to {destination_blob_name} in bucket {bucket_name}.') + + +def generate_and_upload_user_history(execution_date, gcs_bucket_name): + df = download_raw_library_items(execution_date, gcs_bucket_name) + with tempfile.TemporaryDirectory() as tmpdir: + user_preferences = aggregate_user_preferences(df, tmpdir) + dataframes = load_feather_files(tmpdir) + filename = os.path.join(tmpdir, 'user_features.pkl') + save_tables_to_pickle(dataframes, filename) + files = load_tables_from_pickle(filename) + print("GENERATED FEATURE TABLES:", files.keys()) + for table in files.keys(): + print("TABLE: ", table, "LEN: ", len(files[table])) + upload_to_gcs(gcs_bucket_name, filename, f'data/features/user_features.pkl') + + + +def compute_dimension_aggregates(df, dimension, bucket_name): + # Compute initial aggregates to filter out items with less than 2 saved counts + initial_agg = df.groupby(['user_id', dimension]).size().reset_index(name='count') + filtered_df = df[df.set_index(['user_id', dimension]).index.isin(initial_agg[initial_agg['count'] >= 2].set_index(['user_id', dimension]).index)] + + agg = filtered_df.groupby(['user_id', dimension]).agg( + saved_count=(dimension, 'count'), + interaction_count=('user_clicked', 'sum') + ).reset_index() + + agg[f'user_{dimension}_rate_{bucket_name}'] = agg['interaction_count'] / agg['saved_count'] + agg[f'user_{dimension}_proportion_{bucket_name}'] = agg.groupby('user_id')['interaction_count'].transform(lambda x: x / x.sum()) + + agg = agg.rename(columns={ + 'saved_count': f'user_{dimension}_saved_count_{bucket_name}', + 'interaction_count': f'user_{dimension}_interaction_count_{bucket_name}' + }) + + return agg + +def calculate_and_save_aggregates(bucket_name, bucket_df, output_dir): + # Compute aggregates for each dimension + dimensions = ['author', 'site', 'original_url_host', 'subscription'] + for dimension in dimensions: + agg_df = compute_dimension_aggregates(bucket_df, dimension, bucket_name) + + # Save the aggregated DataFrame to a Feather file + filename = os.path.join(output_dir, f'user_{dimension}_{bucket_name}.feather') + save_aggregated_data(agg_df, filename) + print(f"Saved aggregated data for {dimension} in {bucket_name} to {filename}") + + +def save_aggregated_data(df, filename): + buffer = BytesIO() + df.to_feather(buffer) + buffer.seek(0) + + with open(filename, 'wb') as f: + f.write(buffer.getbuffer()) + + +def aggregate_user_preferences(df, output_dir): + # Convert 'created_at' to datetime + df['created_at'] = pd.to_datetime(df['created_at']) + + end_date = df['created_at'].max() + + # Define bucket ranges for the past four weeks + buckets = { + 'week_4': (end_date - timedelta(weeks=4), end_date - timedelta(weeks=3)), + 'week_3': (end_date - timedelta(weeks=3), end_date - timedelta(weeks=2)), + 'week_2': (end_date - timedelta(weeks=2), end_date - timedelta(weeks=1)), + 'week_1': (end_date - timedelta(weeks=1), end_date) + } + + # Calculate aggregates for each bucket and save to file + for bucket_name, (start_date, end_date) in buckets.items(): + bucket_df = df[(df['created_at'] >= start_date) & (df['created_at'] < end_date)] + calculate_and_save_aggregates(bucket_name, bucket_df, output_dir) + + + +def create_and_upload_user_history(execution_date, num_days_history, gcs_bucket_name): + buffer = download_raw_library_items(execution_date, gcs_bucket_name) + buffer = open_raw_library_items() + upload_raw_databuffer(buffer, execution_date, gcs_bucket_name) \ No newline at end of file diff --git a/ml/digest-score/requirements.txt b/ml/digest-score/requirements.txt index b0e2f280e..d62f7cf55 100644 --- a/ml/digest-score/requirements.txt +++ b/ml/digest-score/requirements.txt @@ -6,3 +6,5 @@ google-cloud-storage flask pydantic sklearn2pmml +sqlalchemy +pyarrow diff --git a/ml/digest-score/train.py b/ml/digest-score/train.py index 2789a2221..db8deff84 100644 --- a/ml/digest-score/train.py +++ b/ml/digest-score/train.py @@ -1,20 +1,27 @@ -import psycopg2 import pandas as pd -import joblib -from datetime import datetime - import os -from sklearn.model_selection import train_test_split -from sklearn.preprocessing import StandardScaler -from sklearn.ensemble import RandomForestClassifier -from sklearn2pmml import PMMLPipeline, sklearn2pmml -from sklearn.pipeline import Pipeline -from sklearn.metrics import accuracy_score, classification_report import numpy as np +from datetime import datetime, timedelta + +from sklearn.linear_model import SGDClassifier +from sklearn.ensemble import RandomForestClassifier, VotingClassifier + +from sklearn.preprocessing import StandardScaler + +from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, classification_report, confusion_matrix +from sklearn.utils import shuffle +from sklearn.model_selection import train_test_split +from sklearn2pmml import PMMLPipeline, sklearn2pmml from google.cloud import storage from google.cloud.exceptions import PreconditionFailed +import pickle +import pyarrow as pa +import pyarrow.parquet as pq +import pyarrow.feather as feather + +from features.user_history import FEATURE_COLUMNS DB_PARAMS = { 'dbname': os.getenv('DB_NAME') or 'omnivore', @@ -24,276 +31,175 @@ DB_PARAMS = { 'port': os.getenv('DB_PORT') or '5432' } - -TRAIN_FEATURES = [ - # "item_word_count", - "item_has_thumbnail", - "item_has_site_icon", - - 'user_30d_interactions_author_count', - 'user_30d_interactions_site_count', - 'user_30d_interactions_subscription_count', - - 'user_30d_interactions_author_rate', - 'user_30d_interactions_site_rate', - 'user_30d_interactions_subscription_rate', - - 'global_30d_interactions_site_count', - 'global_30d_interactions_author_count', - 'global_30d_interactions_subscription_count', - - 'global_30d_interactions_site_rate', - 'global_30d_interactions_author_rate', - 'global_30d_interactions_subscription_rate' -] - - -def fetch_data(sample_size): - # Connect to the PostgreSQL database - conn = psycopg2.connect(**DB_PARAMS) - cur = conn.cursor() - query = f""" - SELECT - user_id, - created_at, - item_folder, - item_type, - language, - content_reader, - directionality, - item_word_count, - item_has_thumbnail, - item_has_site_icon, - site, - author, - subscription, - item_subscription_type, - user_clicked, - user_read, - user_long_read - - FROM user_7d_activity LIMIT {sample_size} - """ - - cur.execute(query) - data = cur.fetchall() - - cur.close() - conn.close() - columns = [ - "user_id", - "created_at", - "item_folder", - "item_type", - "language", - "content_reader", - "directionality", - "item_word_count", - "item_has_thumbnail", - "item_has_site_icon", - "site", - "author", - "subscription", - "item_subscription_type", - "user_clicked", - "user_read", - "user_long_read", - ] - - df = pd.DataFrame(data, columns=columns) +def parquet_to_dataframe(file_path): + table = pq.read_table(file_path) + df = table.to_pandas() return df +def save_to_pickle(object, output_file): + with open(output_file, 'wb') as handle: + pickle.dump(object, handle, protocol=pickle.HIGHEST_PROTOCOL) -def add_user_features(df, name, feature_name): - conn = psycopg2.connect(**DB_PARAMS) - cur = conn.cursor() - query = f"SELECT user_id, {name}, interactions, interaction_rate FROM {feature_name}" +def load_tables_from_pickle(pickle_file): + with open(pickle_file, 'rb') as handle: + tables = pickle.load(handle) + return tables - cur.execute(query) - data = cur.fetchall() - - cur.close() - conn.close() - columns = [ - "user_id", - name, - "interactions", - "interaction_rate" - ] - - rate_feature_name = f"{feature_name}_rate" - count_feature_name = f"{feature_name}_count" - - df_loaded = pd.DataFrame(data, columns=columns) - df_loaded = df_loaded.rename(columns={"interactions": count_feature_name}, errors="raise") - df_loaded = df_loaded.rename(columns={"interaction_rate": rate_feature_name}, errors="raise") - df_merged = pd.merge(df, df_loaded[['user_id', name, rate_feature_name, count_feature_name]], on=['user_id',name], how='left') - - df_merged[rate_feature_name] = df_merged[rate_feature_name].fillna(0) - df_merged[count_feature_name] = df_merged[count_feature_name].fillna(0) - - return df_merged +def load_dataframes_from_pickle(pickle_file): + result = {} + tables = load_tables_from_pickle(pickle_file) + for table_name in tables.keys(): + result[table_name] = tables[table_name].to_pandas() + return result -def add_global_features(df, name, feature_name): - conn = psycopg2.connect(**DB_PARAMS) - cur = conn.cursor() - query = f"SELECT {name}, interactions, interaction_rate FROM {feature_name}" - - cur.execute(query) - data = cur.fetchall() - - cur.close() - conn.close() - columns = [ - name, - "interactions", - "interaction_rate" - ] - - rate_feature_name = f"{feature_name}_rate" - count_feature_name = f"{feature_name}_count" - - df_loaded = pd.DataFrame(data, columns=columns) - df_loaded = df_loaded.rename(columns={"interactions": count_feature_name}, errors="raise") - df_loaded = df_loaded.rename(columns={"interaction_rate": rate_feature_name}, errors="raise") - - df_merged = pd.merge(df, df_loaded[[name, count_feature_name, rate_feature_name]], on=name, how='left') - - df_merged[rate_feature_name] = df_merged[rate_feature_name].fillna(0) - df_merged[count_feature_name] = df_merged[count_feature_name].fillna(0) - return df_merged - - -def add_dummy_features(df): - known_folder_types = ['inbox', 'following'] - known_subscription_types = ['NEWSLETTER', 'RSS'] - # known_item_types = ['ARTICLE', 'BOOK', 'FILE', 'HIGHLIGHTS', 'IMAGE', 'PROFILE', 'TWEET', 'UNKNOWN','VIDEO','WEBSITE'] - #known_content_reader_types = ['WEB', 'PDF', 'EPUB'] - # known_directionality_types = ['LTR', 'RTL'] - - folder_dummies = pd.get_dummies(df['item_folder'], columns=known_subscription_types, prefix='item_folder') - subscription_type_dummies = pd.get_dummies(df['item_subscription_type'], columns=known_subscription_types, prefix='item_subscription_type') - - # item_type_dummies = pd.get_dummies(df['item_type'], columns=known_item_types, prefix='item_type') - # content_reader_dummies = pd.get_dummies(df['content_reader'], columns=known_content_reader_types, prefix='content_reader') - # directionality_dummies = pd.get_dummies(df['directionality'], columns=known_directionality_types, prefix='directionality') - # language_dummies = pd.get_dummies(df['language'], prefix='language') - - # if 'title_topic' in df.columns: - # title_topic_dummies = pd.get_dummies(df['title_topic'], prefix='title_topic') - - new_feature_names = list(subscription_type_dummies.columns) + list(folder_dummies.columns) - print("NEW FEATURE NAMES: ", new_feature_names) - # new_feature_names = list(item_type_dummies.columns) + list(content_reader_dummies.columns) + \ - # list(directionality_dummies.columns) + list(language_dummies.columns) - - # if 'title_topic' in df.columns: - # new_feature_names += list(title_topic_dummies.columns) - # , title_topic_dummies - return pd.concat([df, subscription_type_dummies, folder_dummies], axis=1), new_feature_names - - -def random_forest_predictor(df, feature_columns, user_interaction): - features = df[feature_columns] - - features = features.fillna(0) - target = df[user_interaction] - - X = features - y = target.values - X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) - scaler = StandardScaler() - rf_classifier = RandomForestClassifier(n_estimators=50, max_depth=10, random_state=42) - - pipeline = PMMLPipeline([ - ("scaler", scaler), - ("classifier", rf_classifier) - ]) - pipeline.fit(X_train, y_train) - - y_pred = pipeline.predict(X_test) - - feature_importance = rf_classifier.feature_importances_ - - print("Feature Importance:") - for feature, importance in zip(feature_columns, feature_importance): - print(f"{feature}: {importance}") - - print("\nClassification Report:") - print(classification_report(y_test, y_pred)) - - return pipeline - - -def save_and_upload_model(pipeline, target_interaction_type, bucket_name): - pipeline_file_name = f'predict_{target_interaction_type}_random_forest_pipeline-v001.pkl' - joblib.dump(pipeline, pipeline_file_name) - - if bucket_name: - upload_to_gcs(bucket_name, pipeline_file_name, f'models/{pipeline_file_name}') - else: - print("No GCS credentials so i am not uploading") +def download_from_gcs(bucket_name, source_blob_name, destination_file_name): + client = storage.Client() + bucket = client.bucket(bucket_name) + blob = bucket.blob(source_blob_name) + blob.download_to_filename(destination_file_name) + print(f'Blob {source_blob_name} downloaded to {destination_file_name}.') def upload_to_gcs(bucket_name, source_file_name, destination_blob_name): - """Uploads a file to the bucket.""" storage_client = storage.Client() bucket = storage_client.bucket(bucket_name) blob = bucket.blob(destination_blob_name) blob.upload_from_filename(source_file_name) - print(f"File {source_file_name} uploaded to {destination_blob_name}.") -def resample_data(df): - print("Initial distribution:\n", df['user_clicked'].value_counts()) +def load_and_sample_library_items_from_parquet(raw_file_path, sample_size): + df = parquet_to_dataframe(raw_file_path) + sampled_df = df.sample(frac=sample_size, random_state=42) + return sampled_df - # Separate the majority and minority classes - df_majority = df[df['user_clicked'] == False] - df_minority = df[df['user_clicked'] == True] - # Resample the minority class - df_minority_oversampled = df_minority.sample(n=len(df_majority), replace=True, random_state=42) +def merge_user_preference_data(sampled_raw_df, feature_dict): + merged_df = sampled_raw_df - # Combine the majority class with the oversampled minority class - df_balanced = pd.concat([df_majority, df_minority_oversampled]) + for key in feature_dict.keys(): + user_preference_df = feature_dict[key] + if 'author' in key: + merge_keys = ['user_id', 'author'] + elif 'site' in key: + merge_keys = ['user_id', 'site'] + elif 'subscription' in key: + merge_keys = ['user_id', 'subscription'] + elif 'original_url_host' in key: + merge_keys = ['user_id', 'original_url_host'] + else: + print("skipping feature: ", key) + continue # Skip files that don't match expected patterns + merged_df = pd.merge(merged_df, user_preference_df, on=merge_keys, how='left') + merged_df = merged_df.fillna(0) + return merged_df - # Shuffle the DataFrame to mix the classes - df_balanced = df_balanced.sample(frac=1, random_state=42).reset_index(drop=True) +def prepare_data(df): + df['created_at'] = pd.to_datetime(df['created_at']) + df['subscription_start_date'] = pd.to_datetime(df['subscription_start_date'], errors='coerce') - # Check the new distribution - print("Balanced distribution:\n", df_balanced['user_clicked'].value_counts()) + df['is_subscription'] = df['subscription'].apply(lambda x: 1 if pd.notna(x) and x != '' else 0) + df['has_author'] = df['author'].apply(lambda x: 1 if pd.notna(x) and x != '' else 0) - # Display the first few rows of the balanced DataFrame - print(df_balanced.head()) + # Calculate the days since subscribed + df['days_since_subscribed'] = (df['created_at'] - df['subscription_start_date']).dt.days + + # Handle cases where subscription_start_date is NaT (Not a Time) or negative + df['days_since_subscribed'] = df['days_since_subscribed'].apply(lambda x: x if x >= 0 else 0) + df['days_since_subscribed'] = df['days_since_subscribed'].fillna(0).astype(int) + + df['is_feed'] = df['subscription_type'].apply(lambda x: 1 if x == 'RSS' else 0) + df['is_newsletter'] = df['subscription_type'].apply(lambda x: 1 if x == 'NEWSLETTER' else 0) + + df = df.dropna(subset=['user_clicked']) + + # Fill NaNs in other columns with 0 (if any remain) + df = df.fillna(0) + + X = df[FEATURE_COLUMNS] # .drop(columns=['user_id', 'user_clicked']) + Y = df['user_clicked'] + + return X, Y + +def train_random_forest_model(X, Y): + model = RandomForestClassifier( + class_weight={0: 1, 1: 10}, + n_estimators=10, + max_depth=10, + random_state=42 + ) + + scaler = StandardScaler() + X_scaled = scaler.fit_transform(X) + + X_train, X_test, Y_train, Y_test = train_test_split(X_scaled, Y, test_size=0.3, random_state=42) + + pipeline = PMMLPipeline([ + ("scaler", scaler), + ("classifier", model) + ]) + + pipeline.fit(X_train, Y_train) + + Y_pred = pipeline.predict(X_test) + print_classification_report(Y_test, Y_pred) + print_feature_importance(X, model) + + return pipeline + + +def print_feature_importance(X, rf): + # Get feature importances + importances = rf.feature_importances_ + + # Get the indices of the features sorted by importance + indices = np.argsort(importances)[::-1] + + # Print the feature ranking + print("Feature ranking:") + + for f in range(X.shape[1]): + print(f"{f + 1}. feature {indices[f]} ({importances[indices[f]]:.4f}) - {X.columns[indices[f]]}") + + + +def print_classification_report(Y_test, Y_pred): + report = classification_report(Y_test, Y_pred, target_names=['Not Clicked', 'Clicked'], output_dict=True) + print("Classification Report:") + print(f"Accuracy: {report['accuracy']:.4f}") + print(f"Precision (Not Clicked): {report['Not Clicked']['precision']:.4f}") + print(f"Recall (Not Clicked): {report['Not Clicked']['recall']:.4f}") + print(f"F1-Score (Not Clicked): {report['Not Clicked']['f1-score']:.4f}") + print(f"Precision (Clicked): {report['Clicked']['precision']:.4f}") + print(f"Recall (Clicked): {report['Clicked']['recall']:.4f}") + print(f"F1-Score (Clicked): {report['Clicked']['f1-score']:.4f}") - return df_balanced def main(): - sample_size = int(os.getenv('SAMPLE_SIZE')) or 1000 - num_days_history = int(os.getenv('NUM_DAYS_HISTORY')) or 21 - gcs_bucket = os.getenv('GCS_BUCKET') + execution_date = os.getenv('EXECUTION_DATE') + num_days_history = os.getenv('NUM_DAYS_HISTORY') + gcs_bucket_name = os.getenv('GCS_BUCKET') - print("about to fetch library data") - df = fetch_data(sample_size) - print("FETCHED", df) + raw_data_path = f'raw_library_items_${execution_date}.parquet' + user_history_path = 'features_user_features.pkl' + pipeline_path = 'predict_read_pipeline-v002.pkl' - df = add_user_features(df, 'author', 'user_30d_interactions_author') - df = add_user_features(df, 'site', 'user_30d_interactions_site') - df = add_user_features(df, 'subscription', 'user_30d_interactions_subscription') - df = add_global_features(df, 'site', 'global_30d_interactions_site') - df = add_global_features(df, 'author', 'global_30d_interactions_author') - df = add_global_features(df, 'subscription', 'global_30d_interactions_subscription') + download_from_gcs(gcs_bucket_name, f'data/features/user_features.pkl', user_history_path) + download_from_gcs(gcs_bucket_name, f'data/raw/library_items_{execution_date}.parquet', raw_data_path) - df = resample_data(df) - print("training RandomForest with number of library_items: ", len(df)) - pipeline = random_forest_predictor(df, TRAIN_FEATURES, 'user_clicked') + sampled_raw_df = load_and_sample_library_items_from_parquet(raw_data_path, 0.10) + user_history = load_dataframes_from_pickle(user_history_path) - print(f"uploading model and scaler to {gcs_bucket}") - save_and_upload_model(pipeline, 'user_clicked', gcs_bucket) + merged_df = merge_user_preference_data(sampled_raw_df, user_history) + + print("created merged data", merged_df.columns) + + X, Y = prepare_data(merged_df) + random_forest_pipeline = train_random_forest_model(X, Y) + save_to_pickle(random_forest_pipeline, pipeline_path) + upload_to_gcs(gcs_bucket_name, pipeline_path, f'data/models/{pipeline_path}') - print("done") if __name__ == "__main__": main() \ No newline at end of file diff --git a/packages/api/src/data_source.ts b/packages/api/src/data_source.ts index eeb57fd9c..d18eda8f8 100644 --- a/packages/api/src/data_source.ts +++ b/packages/api/src/data_source.ts @@ -24,3 +24,26 @@ export const appDataSource = new DataSource({ idleTimeoutMillis: 10000, // 10 seconds }, }) + +if (env.pg.replication) { + appDataSource.setOptions({ + replication: { + master: { + host: env.pg.host, + port: env.pg.port, + username: env.pg.userName, + password: env.pg.password, + database: env.pg.dbName, + }, + slaves: [ + { + host: env.pg.slave.host, + port: env.pg.slave.port, + username: env.pg.slave.userName, + password: env.pg.slave.password, + database: env.pg.slave.dbName, + }, + ], + }, + }) +} diff --git a/packages/api/src/jobs/score_library_item.ts b/packages/api/src/jobs/score_library_item.ts index 8f084422b..2e89f9955 100644 --- a/packages/api/src/jobs/score_library_item.ts +++ b/packages/api/src/jobs/score_library_item.ts @@ -1,5 +1,10 @@ -import { findLibraryItemById } from '../services/library_item' +import { SubscriptionType } from '../entity/subscription' +import { + findLibraryItemById, + updateLibraryItem, +} from '../services/library_item' import { Feature, scoreClient } from '../services/score' +import { findSubscriptionsByNames } from '../services/subscriptions' import { enqueueUpdateHomeJob } from '../utils/createTask' import { lanaugeToCode } from '../utils/helpers' import { logger } from '../utils/logger' @@ -31,6 +36,8 @@ export const scoreLibraryItem = async ( 'author', 'itemLanguage', 'wordCount', + 'subscription', + 'publishedAt', ], }) if (!libraryItem) { @@ -38,6 +45,26 @@ export const scoreLibraryItem = async ( return } + let subscription + if (libraryItem.subscription) { + const subscriptions = await findSubscriptionsByNames(userId, [ + libraryItem.subscription, + ]) + + if (subscriptions.length) { + subscription = subscriptions[0] + + if (subscription.type === SubscriptionType.Rss) { + logger.info('Skipping scoring for RSS subscription', { + userId, + libraryItemId, + }) + + return + } + } + } + const itemFeatures = { [libraryItem.id]: { library_item_id: libraryItem.id, @@ -53,7 +80,15 @@ export const scoreLibraryItem = async ( language: lanaugeToCode(libraryItem.itemLanguage || 'English'), word_count: libraryItem.wordCount, published_at: libraryItem.publishedAt, - subscription: libraryItem.subscription, + subscription: subscription?.name, + inbox_folder: libraryItem.folder === 'inbox', + is_feed: subscription?.type === SubscriptionType.Rss, + is_newsletter: subscription?.type === SubscriptionType.Newsletter, + is_subscription: !!subscription, + item_word_count: libraryItem.wordCount, + subscription_auto_add_to_library: subscription?.autoAddToLibrary, + subscription_fetch_content: subscription?.fetchContent, + subscription_count: 0, } as Feature, } @@ -69,15 +104,15 @@ export const scoreLibraryItem = async ( throw new Error('Failed to score library item') } - // await updateLibraryItem( - // libraryItem.id, - // { - // score, - // }, - // userId, - // undefined, - // true - // ) + await updateLibraryItem( + libraryItem.id, + { + score, + }, + userId, + undefined, + true + ) logger.info('Library item scored', data) try { diff --git a/packages/api/src/jobs/update_home.ts b/packages/api/src/jobs/update_home.ts index 661b235d0..b565ed7a2 100644 --- a/packages/api/src/jobs/update_home.ts +++ b/packages/api/src/jobs/update_home.ts @@ -1,7 +1,7 @@ import client from 'prom-client' import { LibraryItem } from '../entity/library_item' import { PublicItem } from '../entity/public_item' -import { Subscription } from '../entity/subscription' +import { Subscription, SubscriptionType } from '../entity/subscription' import { User } from '../entity/user' import { registerMetric } from '../prometheus' import { redisDataSource } from '../redis_data_source' @@ -41,6 +41,9 @@ interface Candidate { subscription?: { name: string type: string + autoAddToLibrary?: boolean | null + createdAt: Date + fetchContent?: boolean | null } } @@ -102,6 +105,7 @@ const publicItemToCandidate = (item: PublicItem): Candidate => ({ subscription: { name: item.source.name, type: item.source.type, + createdAt: item.source.createdAt, }, score: 0, }) @@ -222,6 +226,20 @@ const rankCandidates = async ( word_count: item.wordCount, published_at: item.publishedAt, subscription: item.subscription?.name, + inbox_folder: item.folder === 'inbox', + is_feed: item.subscription?.type === SubscriptionType.Rss, + is_newsletter: item.subscription?.type === SubscriptionType.Newsletter, + is_subscription: !!item.subscription, + item_word_count: item.wordCount, + subscription_count: 0, + subscription_auto_add_to_library: item.subscription?.autoAddToLibrary, + subscription_fetch_content: item.subscription?.fetchContent, + days_since_subscribed: item.subscription + ? Math.floor( + (Date.now() - item.subscription.createdAt.getTime()) / + (1000 * 60 * 60 * 24) + ) + : undefined, } as Feature return acc }, {} as Record), diff --git a/packages/api/src/resolvers/function_resolvers.ts b/packages/api/src/resolvers/function_resolvers.ts index febe9deb0..67624303a 100644 --- a/packages/api/src/resolvers/function_resolvers.ts +++ b/packages/api/src/resolvers/function_resolvers.ts @@ -6,7 +6,7 @@ import { createHmac } from 'crypto' import { isError } from 'lodash' import { Highlight } from '../entity/highlight' -import { LibraryItem } from '../entity/library_item' +import { LibraryItem, LibraryItemState } from '../entity/library_item' import { EXISTING_NEWSLETTER_FOLDER, NewsletterEmail, @@ -157,7 +157,7 @@ import { } from './recent_emails' import { recentSearchesResolver } from './recent_searches' import { subscriptionResolver } from './subscriptions' -import { WithDataSourcesContext } from './types' +import { ResolverContext } from './types' import { updateEmailResolver } from './user' /* eslint-disable @typescript-eslint/naming-convention */ @@ -180,7 +180,7 @@ const readingProgressHandlers = { async readingProgressPercent( article: LibraryItem, _: unknown, - ctx: WithDataSourcesContext + ctx: ResolverContext ) { if (ctx.claims?.uid) { const readingProgress = @@ -200,7 +200,7 @@ const readingProgressHandlers = { async readingProgressAnchorIndex( article: LibraryItem, _: unknown, - ctx: WithDataSourcesContext + ctx: ResolverContext ) { if (ctx.claims?.uid) { const readingProgress = @@ -220,7 +220,7 @@ const readingProgressHandlers = { async readingProgressTopPercent( article: LibraryItem, _: unknown, - ctx: WithDataSourcesContext + ctx: ResolverContext ) { if (ctx.claims?.uid) { const readingProgress = @@ -364,11 +364,7 @@ export const functionResolvers = { } return undefined }, - async features( - _: User, - __: Record, - ctx: WithDataSourcesContext - ) { + async features(_: User, __: Record, ctx: ResolverContext) { if (!ctx.claims?.uid) { return undefined } @@ -378,7 +374,7 @@ export const functionResolvers = { async featureList( _: User, __: Record, - ctx: WithDataSourcesContext + ctx: ResolverContext ) { if (!ctx.claims?.uid) { return undefined @@ -398,7 +394,7 @@ export const functionResolvers = { sharedNotesCount: () => 0, }, Article: { - async url(article: LibraryItem, _: unknown, ctx: WithDataSourcesContext) { + async url(article: LibraryItem, _: unknown, ctx: ResolverContext) { if ( (article.itemType == PageType.File || article.itemType == PageType.Book) && @@ -439,20 +435,12 @@ export const functionResolvers = { ? wordsCount(article.readableContent) : undefined }, - async labels( - article: LibraryItem, - _: unknown, - ctx: WithDataSourcesContext - ) { + async labels(article: LibraryItem, _: unknown, ctx: ResolverContext) { if (article.labels) return article.labels return ctx.dataLoaders.labels.load(article.id) }, - async highlights( - article: LibraryItem, - _: unknown, - ctx: WithDataSourcesContext - ) { + async highlights(article: LibraryItem, _: unknown, ctx: ResolverContext) { if (article.highlights) return article.highlights return ctx.dataLoaders.highlights.load(article.id) @@ -468,35 +456,27 @@ export const functionResolvers = { reactions: () => [], replies: () => [], type: (highlight: Highlight) => highlight.highlightType, - async user(highlight: Highlight, __: unknown, ctx: WithDataSourcesContext) { + async user(highlight: Highlight, __: unknown, ctx: ResolverContext) { return ctx.dataLoaders.users.load(highlight.userId) }, - createdByMe( - highlight: Highlight, - __: unknown, - ctx: WithDataSourcesContext - ) { - return highlight.userId === ctx.uid + createdByMe(highlight: Highlight, __: unknown, ctx: ResolverContext) { + return highlight.userId === ctx.claims?.uid }, - libraryItem(highlight: Highlight, _: unknown, ctx: WithDataSourcesContext) { + libraryItem(highlight: Highlight, _: unknown, ctx: ResolverContext) { if (highlight.libraryItem) { return highlight.libraryItem } return ctx.dataLoaders.libraryItems.load(highlight.libraryItemId) }, - labels: async ( - highlight: Highlight, - _: unknown, - ctx: WithDataSourcesContext - ) => { + labels: async (highlight: Highlight, _: unknown, ctx: ResolverContext) => { return ( highlight.labels || ctx.dataLoaders.highlightLabels.load(highlight.id) ) }, }, SearchItem: { - async url(item: LibraryItem, _: unknown, ctx: WithDataSourcesContext) { + async url(item: LibraryItem, _: unknown, ctx: ResolverContext) { if ( (item.itemType == PageType.File || item.itemType == PageType.Book) && ctx.claims && @@ -528,47 +508,33 @@ export const functionResolvers = { return item.siteIcon }, - async labels(item: LibraryItem, _: unknown, ctx: WithDataSourcesContext) { + async labels(item: LibraryItem, _: unknown, ctx: ResolverContext) { if (item.labels) return item.labels return ctx.dataLoaders.labels.load(item.id) }, - async recommendations( - item: LibraryItem, - _: unknown, - ctx: WithDataSourcesContext - ) { + async recommendations(item: LibraryItem, _: unknown, ctx: ResolverContext) { if (item.recommendations) return item.recommendations return ctx.dataLoaders.recommendations.load(item.id) }, - async aiSummary( - item: LibraryItem, - _: unknown, - ctx: WithDataSourcesContext - ) { + async aiSummary(item: LibraryItem, _: unknown, ctx: ResolverContext) { + if (!ctx.claims) return undefined + return ( await getAISummary({ - userId: ctx.uid, + userId: ctx.claims.uid, libraryItemId: item.id, idx: 'latest', }) )?.summary }, - async highlights( - item: LibraryItem, - _: unknown, - ctx: WithDataSourcesContext - ) { + async highlights(item: LibraryItem, _: unknown, ctx: ResolverContext) { if (item.highlights) return item.highlights return ctx.dataLoaders.highlights.load(item.id) }, - async content( - item: PartialLibraryItem, - _: unknown, - ctx: WithDataSourcesContext - ) { + async content(item: PartialLibraryItem, _: unknown, ctx: ResolverContext) { // convert html to the requested format if requested if ( item.format && @@ -658,7 +624,7 @@ export const functionResolvers = { }> }, _: unknown, - ctx: WithDataSourcesContext + ctx: ResolverContext ) { const items = section.items @@ -668,7 +634,14 @@ export const functionResolvers = { const libraryItems = ( await ctx.dataLoaders.libraryItems.loadMany(libraryItemIds) ).filter( - (libraryItem) => !!libraryItem && !isError(libraryItem) + (libraryItem) => + !!libraryItem && + !isError(libraryItem) && + [ + LibraryItemState.Succeeded, + LibraryItemState.ContentNotFetched, + ].includes(libraryItem.state) && + !libraryItem.seenAt ) as Array const publicItemIds = section.items @@ -745,7 +718,7 @@ export const functionResolvers = { { subscription?: string; siteName: string; siteIcon?: string } >, _: unknown, - ctx: WithDataSourcesContext + ctx: ResolverContext ): Promise { if (item.source) { return item.source @@ -785,7 +758,7 @@ export const functionResolvers = { ArticleSavingRequest: { status: (item: LibraryItem) => item.state, url: (item: LibraryItem) => item.originalUrl, - async user(_item: LibraryItem, __: unknown, ctx: WithDataSourcesContext) { + async user(_item: LibraryItem, __: unknown, ctx: ResolverContext) { if (ctx.claims?.uid) { return ctx.dataLoaders.users.load(ctx.claims.uid) } diff --git a/packages/api/src/resolvers/report/index.ts b/packages/api/src/resolvers/report/index.ts index 6ccbf76a5..e4b5eeb90 100644 --- a/packages/api/src/resolvers/report/index.ts +++ b/packages/api/src/resolvers/report/index.ts @@ -11,7 +11,7 @@ import { saveContentDisplayReport, } from '../../services/reports' import { analytics } from '../../utils/analytics' -import { WithDataSourcesContext } from '../types' +import { ResolverContext } from '../types' const SUCCESS_MESSAGE = `Your report has been submitted. Thank you.` const FAILURE_MESSAGE = @@ -36,7 +36,7 @@ const isContentDisplayReport = (types: ReportType[]): boolean => { export const reportItemResolver: ResolverFn< ReportItemResult, unknown, - WithDataSourcesContext, + ResolverContext, MutationReportItemArgs > = async (_obj, args, ctx) => { const { sharedBy, reportTypes } = args.input diff --git a/packages/api/src/resolvers/types.ts b/packages/api/src/resolvers/types.ts index 16857d85d..47fa91954 100644 --- a/packages/api/src/resolvers/types.ts +++ b/packages/api/src/resolvers/types.ts @@ -14,7 +14,6 @@ import { Recommendation } from '../entity/recommendation' import { Subscription } from '../entity/subscription' import { UploadFile } from '../entity/upload_file' import { User } from '../entity/user' -import { HomeItem } from '../generated/graphql' import { PubsubClient } from '../pubsub' export interface Claims { @@ -65,7 +64,3 @@ export interface RequestContext { } export type ResolverContext = ApolloContext - -export type WithDataSourcesContext = { - uid: string -} & ResolverContext diff --git a/packages/api/src/resolvers/user/index.ts b/packages/api/src/resolvers/user/index.ts index 6572c1f10..9c496df8c 100644 --- a/packages/api/src/resolvers/user/index.ts +++ b/packages/api/src/resolvers/user/index.ts @@ -45,7 +45,7 @@ import { softDeleteUser } from '../../services/user' import { Merge } from '../../util' import { authorized } from '../../utils/gql-utils' import { validateUsername } from '../../utils/usernamePolicy' -import { WithDataSourcesContext } from '../types' +import { ResolverContext } from '../types' export const updateUserResolver = authorized< Merge, @@ -145,7 +145,7 @@ export const updateUserProfileResolver = authorized< export const googleLoginResolver: ResolverFn< Merge, unknown, - WithDataSourcesContext, + ResolverContext, MutationGoogleLoginArgs > = async (_obj, { input }, { setAuth }) => { const { email, secret } = input @@ -172,7 +172,7 @@ export const googleLoginResolver: ResolverFn< export const validateUsernameResolver: ResolverFn< boolean, Record, - WithDataSourcesContext, + ResolverContext, QueryValidateUsernameArgs > = async (_obj, { username }) => { const lowerCasedUsername = username.toLowerCase() @@ -191,7 +191,7 @@ export const validateUsernameResolver: ResolverFn< export const googleSignupResolver: ResolverFn< Merge, Record, - WithDataSourcesContext, + ResolverContext, MutationGoogleSignupArgs > = async (_obj, { input }, { setAuth, log }) => { const { email, username, name, bio, sourceUserId, pictureUrl, secret } = input @@ -231,7 +231,7 @@ export const googleSignupResolver: ResolverFn< export const logOutResolver: ResolverFn< LogOutResult, unknown, - WithDataSourcesContext, + ResolverContext, unknown > = (_, __, { clearAuth, log }) => { try { @@ -246,7 +246,7 @@ export const logOutResolver: ResolverFn< export const getMeUserResolver: ResolverFn< UserEntity | undefined, unknown, - WithDataSourcesContext, + ResolverContext, unknown > = async (_obj, __, { claims }) => { try { @@ -268,9 +268,9 @@ export const getMeUserResolver: ResolverFn< export const getUserResolver: ResolverFn< Merge, unknown, - WithDataSourcesContext, + ResolverContext, QueryUserArgs -> = async (_obj, { userId: id, username }, { uid }) => { +> = async (_obj, { userId: id, username }) => { if (!(id || username)) { return { errorCodes: [UserErrorCode.BadRequest] } } diff --git a/packages/api/src/resolvers/user_feed_article/index.ts b/packages/api/src/resolvers/user_feed_article/index.ts deleted file mode 100644 index 928078f9d..000000000 --- a/packages/api/src/resolvers/user_feed_article/index.ts +++ /dev/null @@ -1,150 +0,0 @@ -/* eslint-disable @typescript-eslint/require-await */ -/* eslint-disable @typescript-eslint/no-unused-vars */ -import { FeedArticle, PageInfo } from '../../generated/graphql' - -export type PartialFeedArticle = Omit< - FeedArticle, - 'sharedBy' | 'article' | 'reactions' -> - -type PaginatedFeedArticlesSuccessPartial = { - edges: { cursor: string; node: PartialFeedArticle }[] - pageInfo: PageInfo -} - -// export const getSharedArticleResolver: ResolverFn< -// SharedArticleSuccessPartial | SharedArticleError, -// Record, -// WithDataSourcesContext, -// QuerySharedArticleArgs -// > = async (_obj, { username, slug, selectedHighlightId }, { kx, models }) => { -// try { -// const user = await models.user.getWhere({ username }) -// if (!user) { -// return { -// errorCodes: [SharedArticleErrorCode.NotFound], -// } -// } - -// const article = await models.userArticle.getBySlug(username, slug) -// if (!article || !article.sharedAt) { -// return { -// errorCodes: [SharedArticleErrorCode.NotFound], -// } -// } - -// if (selectedHighlightId) { -// const highlightResult = await models.highlight.getWhereIn('shortId', [ -// selectedHighlightId, -// ]) -// if (!highlightResult || !highlightResult[0].sharedAt) { -// return { -// errorCodes: [SharedArticleErrorCode.NotFound], -// } -// } -// } - -// const shareInfo = await getShareInfoForArticle( -// kx, -// user.id, -// article.id, -// models -// ) - -// return { article: { ...article, userId: user.id, shareInfo: shareInfo } } -// } catch (error) { -// return { errorCodes: [SharedArticleErrorCode.NotFound] } -// } -// } - -// export const getUserFeedArticlesResolver: ResolverFn< -// PaginatedFeedArticlesSuccessPartial, -// unknown, -// WithDataSourcesContext, -// QueryFeedArticlesArgs -// > = async ( -// _obj, -// { after: _startCursor, first: _first, sharedByUser }, -// { models, claims, authTrx } -// ) => { -// if (!(sharedByUser || claims?.uid)) { -// return { -// edges: [], -// pageInfo: { -// startCursor: '', -// endCursor: '', -// hasNextPage: false, -// hasPreviousPage: false, -// }, -// } -// } - -// const first = _first || 0 -// const startCursor = _startCursor || '' - -// const feedArticles = -// (await authTrx((tx) => -// models.userArticle.getUserFeedArticlesPaginatedWithHighlights( -// { cursor: startCursor, first: first + 1, sharedByUser }, // fetch one more item to get next cursor -// claims?.uid || '', -// tx -// ) -// )) || [] - -// const endCursor = feedArticles[feedArticles.length - 1]?.sharedAt -// .getTime() -// ?.toString() -// const hasNextPage = feedArticles.length > first - -// if (hasNextPage) { -// // remove an extra if exists -// feedArticles.pop() -// } - -// const edges = feedArticles.map((fa) => { -// return { -// node: fa, -// cursor: fa.sharedAt.getTime()?.toString(), -// } -// }) - -// return { -// edges, -// pageInfo: { -// hasPreviousPage: false, -// startCursor: '', -// hasNextPage, -// endCursor, -// }, -// } -// } - -// export const updateSharedCommentResolver = authorized< -// UpdateSharedCommentSuccess, -// UpdateSharedCommentError, -// MutationUpdateSharedCommentArgs -// >( -// async ( -// _, -// { input: { articleID, sharedComment } }, -// { models, authTrx, claims: { uid } } -// ) => { -// const ua = await authTrx((tx) => -// models.userArticle.getByParameters(uid, { articleId: articleID }, tx) -// ) -// if (!ua) { -// return { errorCodes: [UpdateSharedCommentErrorCode.NotFound] } -// } - -// await authTrx((tx) => -// models.userArticle.updateByArticleId( -// uid, -// articleID, -// { sharedComment }, -// tx -// ) -// ) - -// return { articleID, sharedComment } -// } -// ) diff --git a/packages/api/src/resolvers/user_friends/index.ts b/packages/api/src/resolvers/user_friends/index.ts deleted file mode 100644 index 823184066..000000000 --- a/packages/api/src/resolvers/user_friends/index.ts +++ /dev/null @@ -1,101 +0,0 @@ -// export const setFollowResolver = authorized< -// SetFollowSuccess, -// SetFollowError, -// MutationSetFollowArgs -// >( -// async ( -// _, -// { input: { userId: friendUserId, follow } }, -// { models, authTrx, claims: { uid } } -// ) => { -// const user = await models.user.getUserDetails(uid, friendUserId) -// if (!user) return { errorCodes: [SetFollowErrorCode.NotFound] } - -// const userFriendRecord = await authTrx((tx) => -// models.userFriends.getByUserFriendId(uid, friendUserId, tx) -// ) - -// if (follow) { -// if (!userFriendRecord) { -// await authTrx((tx) => -// models.userFriends.create({ friendUserId, userId: uid }, tx) -// ) -// } -// } else if (userFriendRecord) { -// await authTrx((tx) => models.userFriends.delete(userFriendRecord.id, tx)) -// } - -// const updatedUser = await models.user.getUserDetails(uid, friendUserId) -// if (!updatedUser) return { errorCodes: [SetFollowErrorCode.NotFound] } - -// return { -// updatedUser: { -// ...userDataToUser(updatedUser), -// isFriend: updatedUser.viewerIsFollowing, -// }, -// } -// } -// ) - -// const getUserList = async ( -// uid: string, -// users: UserData[], -// models: DataModels, -// authTrx: ( -// cb: (tx: Knex.Transaction) => TResult, -// userRole?: string -// ) => Promise -// ): Promise => { -// const usersIds = users.map(({ id }) => id) -// const friends = await authTrx((tx) => -// models.userFriends.getByFriendIds(uid, usersIds, tx) -// ) - -// const friendsIds = friends.map(({ friendUserId }) => friendUserId) -// users = users.map((f) => ({ -// ...f, -// isFriend: friendsIds.includes(f.id), -// viewerIsFollowing: friendsIds.includes(f.id), -// })) - -// return users.map((u) => userDataToUser(u)) -// } - -// export const getFollowersResolver: ResolverFn< -// GetFollowersResult, -// unknown, -// WithDataSourcesContext, -// QueryGetFollowersArgs -// > = async (_parent, { userId }, { models, claims, authTrx }) => { -// const followers = userId -// ? await authTrx((tx) => models.user.getUserFollowersList(userId, tx)) -// : [] -// if (!claims?.uid) return { followers: usersWithNoFriends(followers) } -// return { -// followers: await getUserList(claims?.uid, followers, models, authTrx), -// } -// } - -// export const getFollowingResolver: ResolverFn< -// GetFollowingResult, -// unknown, -// WithDataSourcesContext, -// QueryGetFollowingArgs -// > = async (_parent, { userId }, { models, claims, authTrx }) => { -// const following = userId -// ? await authTrx((tx) => models.user.getUserFollowingList(userId, tx)) -// : [] -// if (!claims?.uid) return { following: usersWithNoFriends(following) } -// return { -// following: await getUserList(claims?.uid, following, models, authTrx), -// } -// } - -// const usersWithNoFriends = (users: UserData[]): User[] => { -// return users.map((f) => -// userDataToUser({ -// ...f, -// isFriend: false, -// } as UserData) -// ) -// } diff --git a/packages/api/src/services/highlights.ts b/packages/api/src/services/highlights.ts index 327038aca..533d29eb2 100644 --- a/packages/api/src/services/highlights.ts +++ b/packages/api/src/services/highlights.ts @@ -26,7 +26,6 @@ export const batchGetHighlightsFromLibraryItemIds = async ( const highlights = await authTrx(async (tx) => tx.getRepository(Highlight).find({ where: { libraryItem: { id: In(libraryItemIds as string[]) } }, - relations: ['user'], }) ) diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts index 4ed0136a6..4eee6af0c 100644 --- a/packages/api/src/services/library_item.ts +++ b/packages/api/src/services/library_item.ts @@ -6,7 +6,6 @@ import { EntityManager, FindOptionsWhere, In, - IsNull, ObjectLiteral, } from 'typeorm' import { QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialEntity' @@ -135,31 +134,15 @@ export enum SortBy { const readingProgressDataSource = new ReadingProgressDataSource() export const batchGetLibraryItems = async (ids: readonly string[]) => { - const selectColumns: Array = [ - 'id', - 'title', - 'author', - 'thumbnail', - 'wordCount', - 'savedAt', - 'originalUrl', - 'directionality', - 'description', - 'subscription', - 'siteName', - 'siteIcon', - 'archivedAt', - 'deletedAt', - 'slug', - 'previewContent', - ] + // select all columns except content + const select = getColumns(libraryItemRepository).filter( + (select) => ['originalContent', 'readableContent'].indexOf(select) === -1 + ) const items = await authTrx(async (tx) => tx.getRepository(LibraryItem).find({ - select: selectColumns, + select, where: { id: In(ids as string[]), - state: LibraryItemState.Succeeded, - seenAt: IsNull(), }, }) ) diff --git a/packages/api/src/services/score.ts b/packages/api/src/services/score.ts index c8c8c1660..8752e4b4a 100644 --- a/packages/api/src/services/score.ts +++ b/packages/api/src/services/score.ts @@ -6,6 +6,12 @@ export interface Feature { has_thumbnail: boolean has_site_icon: boolean saved_at: Date + item_word_count: number + is_subscription: boolean + inbox_folder: boolean + is_newsletter: boolean + is_feed: boolean + site?: string language?: string author?: string @@ -15,6 +21,10 @@ export interface Feature { folder?: string published_at?: Date subscription?: string + subscription_auto_add_to_library?: boolean + subscription_fetch_content?: boolean + days_since_subscribed?: number + subscription_count?: number } export interface ScoreApiRequestBody { @@ -69,4 +79,4 @@ class ScoreClientImpl implements ScoreClient { } } -export const scoreClient = new StubScoreClientImpl() +export const scoreClient = new ScoreClientImpl() diff --git a/packages/api/src/util.ts b/packages/api/src/util.ts index 096acf0a0..4e32d5406 100755 --- a/packages/api/src/util.ts +++ b/packages/api/src/util.ts @@ -19,6 +19,14 @@ export interface BackendEnv { pool: { max: number } + replication: boolean + slave: { + host: string + port: number + userName: string + password: string + dbName: string + } } server: { jwtSecret: string @@ -179,6 +187,12 @@ const nullableEnvVars = [ 'NOTION_CLIENT_SECRET', 'NOTION_AUTH_URL', 'SCORE_API_URL', + 'PG_REPLICATION', + 'PG_SLAVE_HOST', + 'PG_SLAVE_PORT', + 'PG_SLAVE_USER', + 'PG_SLAVE_PASSWORD', + 'PG_SLAVE_DB', ] // Allow some vars to be null/empty const envParser = @@ -218,6 +232,14 @@ export function getEnv(): BackendEnv { pool: { max: parseInt(parse('PG_POOL_MAX'), 10), }, + replication: parse('PG_REPLICATION') === 'true', + slave: { + host: parse('PG_SLAVE_HOST'), + port: parseInt(parse('PG_SLAVE_PORT'), 10), + userName: parse('PG_SLAVE_USER'), + password: parse('PG_SLAVE_PASSWORD'), + dbName: parse('PG_SLAVE_DB'), + }, } const server = { jwtSecret: parse('JWT_SECRET'), diff --git a/packages/api/src/utils/gql-utils.ts b/packages/api/src/utils/gql-utils.ts index ef5986f28..62bbf2d09 100644 --- a/packages/api/src/utils/gql-utils.ts +++ b/packages/api/src/utils/gql-utils.ts @@ -1,5 +1,5 @@ import { ResolverFn } from '../generated/graphql' -import { Claims, WithDataSourcesContext } from '../resolvers/types' +import { Claims, ResolverContext } from '../resolvers/types' export function authorized< TSuccess, @@ -12,10 +12,10 @@ export function authorized< resolver: ResolverFn< TSuccess | TError, TParent, - WithDataSourcesContext & { claims: Claims }, + ResolverContext & { claims: Claims; uid: string }, TArgs > -): ResolverFn { +): ResolverFn { return (parent, args, ctx, info) => { const { claims } = ctx if (claims?.uid) { diff --git a/packages/db/migrations/0183.do.alter_omnivore_admin_role.sql b/packages/db/migrations/0183.do.alter_omnivore_admin_role.sql new file mode 100755 index 000000000..ab74e309e --- /dev/null +++ b/packages/db/migrations/0183.do.alter_omnivore_admin_role.sql @@ -0,0 +1,35 @@ +-- Type: DO +-- Name: alter_omnivore_admin_role +-- Description: Alter omnivore_admin role to prevent omnivore_admin to be inherited by app_user or omnivore_user + +BEGIN; + +DROP POLICY user_admin_policy ON omnivore.user; +DROP POLICY library_item_admin_policy ON omnivore.library_item; + +REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA omnivore from omnivore_admin; +REVOKE ALL PRIVILEGES ON SCHEMA omnivore from omnivore_admin; + +DROP ROLE omnivore_admin; + +CREATE ROLE omnivore_admin; + +GRANT USAGE ON SCHEMA omnivore TO omnivore_admin; + +ALTER ROLE omnivore_user NOINHERIT; -- This is to prevent omnivore_user from inheriting omnivore_admin role + +GRANT omnivore_admin TO omnivore_user; -- This is to allow app_user to set omnivore_admin role + +GRANT SELECT, INSERT, UPDATE, DELETE ON omnivore.user TO omnivore_admin; +CREATE POLICY user_admin_policy on omnivore.user + FOR ALL + TO omnivore_admin + USING (true); + +GRANT SELECT, INSERT, UPDATE, DELETE ON omnivore.library_item TO omnivore_admin; +CREATE POLICY library_item_admin_policy ON omnivore.library_item + FOR ALL + TO omnivore_admin + USING (true); + +COMMIT; diff --git a/packages/db/migrations/0183.undo.alter_omnivore_admin_role.sql b/packages/db/migrations/0183.undo.alter_omnivore_admin_role.sql new file mode 100755 index 000000000..2299875d5 --- /dev/null +++ b/packages/db/migrations/0183.undo.alter_omnivore_admin_role.sql @@ -0,0 +1,36 @@ +-- Type: UNDO +-- Name: alter_omnivore_admin_role +-- Description: Alter omnivore_admin role to prevent omnivore_admin to be inherited by app_user or omnivore_user + +BEGIN; + +DROP POLICY library_item_admin_policy ON omnivore.library_item; +REVOKE SELECT, INSERT, UPDATE, DELETE ON omnivore.library_item FROM omnivore_admin; + +DROP POLICY user_admin_policy ON omnivore.user; +REVOKE SELECT, INSERT, UPDATE, DELETE ON omnivore.user FROM omnivore_admin; + +REVOKE USAGE ON SCHEMA omnivore FROM omnivore_admin; + +DROP ROLE omnivore_admin; + +ALTER ROLE omnivore_user INHERIT; + +CREATE ROLE omnivore_admin; + +GRANT omnivore_admin TO app_user; + +GRANT ALL PRIVILEGES ON SCHEMA omnivore TO omnivore_admin; +GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA omnivore TO omnivore_admin; + +CREATE POLICY user_admin_policy on omnivore.user + FOR ALL + TO omnivore_admin + USING (true); + +CREATE POLICY library_item_admin_policy on omnivore.library_item + FOR ALL + TO omnivore_admin + USING (true); + +COMMIT; diff --git a/packages/web/components/elements/ModalPrimitives.tsx b/packages/web/components/elements/ModalPrimitives.tsx index 465a97c9c..2b8a8deeb 100644 --- a/packages/web/components/elements/ModalPrimitives.tsx +++ b/packages/web/components/elements/ModalPrimitives.tsx @@ -35,6 +35,7 @@ const Modal = styled(Content, { export const ModalContent = styled(Modal, { top: '50%', left: '50%', + bg: '$readerBg', transform: 'translate(-50%, -50%)', width: '90vw', maxWidth: '450px', diff --git a/packages/web/components/elements/TickedRangeSlider.tsx b/packages/web/components/elements/TickedRangeSlider.tsx index da7696569..b4119e1ce 100644 --- a/packages/web/components/elements/TickedRangeSlider.tsx +++ b/packages/web/components/elements/TickedRangeSlider.tsx @@ -22,14 +22,14 @@ const StyledSlider = styled(Slider, { height: '8px', width: '225px', borderRadius: '10px', - backgroundColor: '#F2F2F2', + backgroundColor: '$thTextSubtle2', }, '.SliderThumb': { display: 'block', - width: '20px', - height: '20px', + width: '15px', + height: '15px', borderRadius: '50%', - border: '4px solid white', + border: '2px solid $thTextSubtle2', backgroundColor: '#FFD234', boxShadow: '0px 0px 20px rgba(19, 56, 77, 0.2)', }, diff --git a/packages/web/components/nav-containers/highlights.tsx b/packages/web/components/nav-containers/HighlightsContainer.tsx similarity index 100% rename from packages/web/components/nav-containers/highlights.tsx rename to packages/web/components/nav-containers/HighlightsContainer.tsx diff --git a/packages/web/components/nav-containers/home.tsx b/packages/web/components/nav-containers/HomeContainer.tsx similarity index 87% rename from packages/web/components/nav-containers/home.tsx rename to packages/web/components/nav-containers/HomeContainer.tsx index 1174b3334..2dbd80152 100644 --- a/packages/web/components/nav-containers/home.tsx +++ b/packages/web/components/nav-containers/HomeContainer.tsx @@ -27,6 +27,7 @@ import { import { Box, HStack, SpanBox, VStack } from '../elements/LayoutPrimitives' import { Toaster } from 'react-hot-toast' import { useGetViewerQuery } from '../../lib/networking/queries/useGetViewerQuery' +import useLibraryItemActions from '../../lib/hooks/useLibraryItemActions' export function HomeContainer(): JSX.Element { const router = useRouter() @@ -40,7 +41,7 @@ export function HomeContainer(): JSX.Element { }, [viewerData]) useEffect(() => { - window.sessionStorage.setItem('nav-return', router.asPath) + window.localStorage.setItem('nav-return', router.asPath) }, [router.asPath]) return ( @@ -74,7 +75,8 @@ export function HomeContainer(): JSX.Element { > {homeData.sections?.map((homeSection, idx) => { if (homeSection.items.length < 1) { - return <> + console.log('empty home section: ', homeSection) + return } switch (homeSection.layout) { case 'just_added': @@ -110,7 +112,8 @@ export function HomeContainer(): JSX.Element { /> ) default: - return <> + console.log('unknown home section: ', homeSection) + return } })} @@ -151,6 +154,12 @@ const JustAddedHomeSection = (props: HomeSectionProps): JSX.Element => { fontSize: '16px', fontWeight: '600', color: '$homeTextTitle', + overflow: 'hidden', + textOverflow: 'ellipsis', + wordBreak: 'break-word', + display: '-webkit-box', + '-webkit-line-clamp': '2', + '-webkit-box-orient': 'vertical', }} > {props.homeSection.title} @@ -271,7 +280,7 @@ const TopPicksHomeSection = (props: HomeSectionProps): JSX.Element => { ( { ) } -const TitleSmall = (props: HomeItemViewProps): JSX.Element => { +type TitleSmallProps = { + maxLines?: string +} + +const TitleSmall = ( + props: HomeItemViewProps & TitleSmallProps +): JSX.Element => { return ( { textOverflow: 'ellipsis', wordBreak: 'break-word', display: '-webkit-box', - '-webkit-line-clamp': '3', + '-webkit-line-clamp': props.maxLines ?? '3', '-webkit-box-orient': 'vertical', }} > @@ -527,7 +542,7 @@ const JustAddedItemView = (props: HomeItemViewProps): JSX.Element => { bg: '$homeCardHover', borderRadius: '5px', '&:hover': { - bg: '$homeCardHover', + bg: '#007AFF10', }, '&:hover .title-text': { textDecoration: 'underline', @@ -556,7 +571,7 @@ const JustAddedItemView = (props: HomeItemViewProps): JSX.Element => { - + ) } @@ -569,6 +584,9 @@ const TopPicksItemView = ( props: HomeItemViewProps & TopPicksItemViewProps ): JSX.Element => { const router = useRouter() + const { archiveItem, deleteItem, moveItem, shareItem } = + useLibraryItemActions() + return ( {props.homeItem.canSave && ( - )} - {/* */} - {props.homeItem.canArchive && ( )} {props.homeItem.canShare && ( - )} @@ -828,7 +895,7 @@ const SubscriptionSourceHoverContent = ( {props.source.icon && } {subscription && subscription.status == 'ACTIVE' && ( )} diff --git a/packages/web/components/patterns/LibraryCards/LibraryGridCard.tsx b/packages/web/components/patterns/LibraryCards/LibraryGridCard.tsx index 857816954..8bba60f0b 100644 --- a/packages/web/components/patterns/LibraryCards/LibraryGridCard.tsx +++ b/packages/web/components/patterns/LibraryCards/LibraryGridCard.tsx @@ -95,7 +95,7 @@ export function LibraryGridCard(props: LinkedItemCardProps): JSX.Element { props.setIsChecked(props.item.id, !props.isChecked) return } - window.sessionStorage.setItem('nav-return', router.asPath) + window.localStorage.setItem('nav-return', router.asPath) if (event.metaKey || event.ctrlKey) { window.open( `/${props.viewer.profile.username}/${props.item.slug}`, diff --git a/packages/web/components/patterns/LibraryCards/LibraryListCard.tsx b/packages/web/components/patterns/LibraryCards/LibraryListCard.tsx index a7d277230..5fb5d00ab 100644 --- a/packages/web/components/patterns/LibraryCards/LibraryListCard.tsx +++ b/packages/web/components/patterns/LibraryCards/LibraryListCard.tsx @@ -100,7 +100,7 @@ export function LibraryListCard(props: LinkedItemCardProps): JSX.Element { props.setIsChecked(props.item.id, !props.isChecked) return } - window.sessionStorage.setItem('nav-return', router.asPath) + window.localStorage.setItem('nav-return', router.asPath) if (event.metaKey || event.ctrlKey) { window.open( `/${props.viewer.profile.username}/${props.item.slug}`, diff --git a/packages/web/components/templates/SettingsLayout.tsx b/packages/web/components/templates/SettingsLayout.tsx index a6716af9d..f1eb850cf 100644 --- a/packages/web/components/templates/SettingsLayout.tsx +++ b/packages/web/components/templates/SettingsLayout.tsx @@ -32,7 +32,7 @@ const ReturnButton = (): JSX.Element => { }, }} > - + (null) const [notebookKey, setNotebookKey] = useState(uuidv4()) const [noteTarget, setNoteTarget] = useState(undefined) - const [noteTargetPageIndex, setNoteTargetPageIndex] = useState< - number | undefined - >(undefined) + const [noteTargetPageIndex, setNoteTargetPageIndex] = + useState(undefined) const highlightsRef = useRef([]) const annotationOmnivoreId = (annotation: Annotation): string | undefined => { @@ -449,16 +448,16 @@ export default function PdfArticleContainer( document.dispatchEvent(new Event('openOriginalArticle')) break case 'u': - const navReturn = window.sessionStorage.getItem('nav-return') + const navReturn = window.localStorage.getItem('nav-return') if (navReturn) { window.location.assign(navReturn) return } const query = window.sessionStorage.getItem('q') if (query) { - window.location.assign(`/home?${query}`) + window.location.assign(`/l/home?${query}`) } else { - window.location.replace(`/home`) + window.location.replace(`/l/home`) } break case 'e': diff --git a/packages/web/components/tokens/stitches.config.ts b/packages/web/components/tokens/stitches.config.ts index 4351694c9..d07b70932 100644 --- a/packages/web/components/tokens/stitches.config.ts +++ b/packages/web/components/tokens/stitches.config.ts @@ -416,7 +416,7 @@ const blackThemeSpec = { const apolloThemeSpec = { colors: { - readerBg: '#6A6968', + readerBg: '#474747', readerFont: '#F3F3F3', readerMargin: '#474747', readerFontHighContrast: 'white', @@ -431,7 +431,7 @@ const apolloThemeSpec = { homeCardHover: '#525252', homeDivider: '#6A6968', - homeActionHoverBg: '#515151', + homeActionHoverBg: '#474747', thBackground: '#474747', thBackground2: '#515151', diff --git a/packages/web/lib/hooks/useLibraryItemActions.tsx b/packages/web/lib/hooks/useLibraryItemActions.tsx new file mode 100644 index 000000000..8be217b34 --- /dev/null +++ b/packages/web/lib/hooks/useLibraryItemActions.tsx @@ -0,0 +1,91 @@ +import { useState, useEffect, useCallback } from 'react' +import { setLinkArchivedMutation } from '../networking/mutations/setLinkArchivedMutation' +import { + showErrorToast, + showSuccessToast, + showSuccessToastWithUndo, +} from '../toastHelpers' +import { deleteLinkMutation } from '../networking/mutations/deleteLinkMutation' +import { updatePageMutation } from '../networking/mutations/updatePageMutation' +import { State } from '../networking/fragments/articleFragment' + +export default function useLibraryItemActions() { + const archiveItem = useCallback(async (itemId: string) => { + const result = await setLinkArchivedMutation({ + linkId: itemId, + archived: true, + }) + + if (result) { + showSuccessToast('Link archived', { position: 'bottom-right' }) + } else { + showErrorToast('Error archiving link', { position: 'bottom-right' }) + } + + return !!result + }, []) + + const deleteItem = useCallback(async (itemId: string, undo: () => void) => { + const result = await deleteLinkMutation(itemId) + + if (result) { + showSuccessToastWithUndo('Item removed', async () => { + const result = await updatePageMutation({ + pageId: itemId, + state: State.SUCCEEDED, + }) + + undo() + + if (result) { + showSuccessToast('Item recovered') + } else { + showErrorToast('Error recovering, check your deleted items') + } + }) + } else { + showErrorToast('Error removing item', { position: 'bottom-right' }) + } + + return !!result + }, []) + + const moveItem = useCallback(async (itemId: string) => { + const result = await setLinkArchivedMutation({ + linkId: itemId, + archived: true, + }) + + if (result) { + showSuccessToast('Link archived', { position: 'bottom-right' }) + } else { + showErrorToast('Error archiving link', { position: 'bottom-right' }) + } + + return !!result + }, []) + + const shareItem = useCallback( + async (title: string, originalArticleUrl: string | undefined) => { + if (!originalArticleUrl) { + showErrorToast('Article has no public URL to share', { + position: 'bottom-right', + }) + } else if (navigator.share) { + navigator.share({ + title: title + '\n', + text: title + '\n', + url: originalArticleUrl, + }) + } else { + await navigator.clipboard.writeText(originalArticleUrl) + showSuccessToast('URL copied to clipboard', { + position: 'bottom-right', + }) + } + }, + [] + ) + + return { archiveItem, deleteItem, moveItem, shareItem } +} diff --git a/packages/web/lib/networking/mutations/moveToLibraryMutation.ts b/packages/web/lib/networking/mutations/moveToLibraryMutation.ts new file mode 100644 index 000000000..205312dd6 --- /dev/null +++ b/packages/web/lib/networking/mutations/moveToLibraryMutation.ts @@ -0,0 +1,41 @@ +import { gql } from 'graphql-request' +import { gqlFetcher } from '../networkHelpers' + +type MoveToFolderResponseData = { + success?: boolean + errorCodes?: string[] +} + +type MoveToFolderResponse = { + moveToFolder?: MoveToFolderResponseData +} + +export async function moveToFolderMutation( + itemId: string, + folder: string +): Promise { + const mutation = gql` + mutation MoveToFolder($id: ID!, $folder: String!) { + moveToFolder(id: $id, folder: $folder) { + ... on MoveToFolderSuccess { + success + } + ... on MoveToFolderError { + errorCodes + } + } + } + ` + + try { + const response = await gqlFetcher(mutation, { id: itemId, folder }) + const data = response as MoveToFolderResponse | undefined + if (data?.moveToFolder?.errorCodes) { + return false + } + return data?.moveToFolder?.success ?? false + } catch (error) { + console.log('MoveToFolder error', error) + return false + } +} diff --git a/packages/web/pages/[username]/[slug]/index.tsx b/packages/web/pages/[username]/[slug]/index.tsx index 45d2f94c4..559f4f8d0 100644 --- a/packages/web/pages/[username]/[slug]/index.tsx +++ b/packages/web/pages/[username]/[slug]/index.tsx @@ -89,7 +89,7 @@ export default function Home(): JSX.Element { // return // } // } - const navReturn = window.sessionStorage.getItem('nav-return') + const navReturn = window.localStorage.getItem('nav-return') if (navReturn) { router.push(navReturn) return @@ -303,7 +303,7 @@ export default function Home(): JSX.Element { name: 'Return to library', shortcut: ['u'], perform: () => { - const navReturn = window.sessionStorage.getItem('nav-return') + const navReturn = window.localStorage.getItem('nav-return') if (navReturn) { router.push(navReturn) return diff --git a/packages/web/pages/_app.tsx b/packages/web/pages/_app.tsx index e81880576..86f82fcce 100644 --- a/packages/web/pages/_app.tsx +++ b/packages/web/pages/_app.tsx @@ -43,12 +43,12 @@ const generateActions = (router: NextRouter) => { shortcut: ['g', 'h'], keywords: 'go home', perform: () => { - const navReturn = window.sessionStorage.getItem('nav-return') + const navReturn = window.localStorage.getItem('nav-return') if (navReturn) { router.push(navReturn) return } - router?.push('/home') + router?.push('/l/home') }, }, { diff --git a/packages/web/pages/api/client/auth.ts b/packages/web/pages/api/client/auth.ts index ee1d2dc01..e63d81767 100644 --- a/packages/web/pages/api/client/auth.ts +++ b/packages/web/pages/api/client/auth.ts @@ -29,7 +29,7 @@ const requestHandler = (req: NextApiRequest, res: NextApiResponse): void => { }) } else { res.writeHead(302, { - Location: '/home', + Location: '/l/home', }) } diff --git a/packages/web/pages/index.tsx b/packages/web/pages/index.tsx index 199614dc2..51b2b3b14 100644 --- a/packages/web/pages/index.tsx +++ b/packages/web/pages/index.tsx @@ -9,7 +9,12 @@ export default function LandingPage(): JSX.Element { const { viewerData, isLoading } = useGetViewerQuery() if (!isLoading && router.isReady && viewerData?.me) { - router.push('/home') + const navReturn = window.localStorage.getItem('nav-return') + if (navReturn) { + router.push(navReturn) + } else { + router.push('/l/home') + } return <> } else if (isLoading || !router.isReady) { return ( diff --git a/packages/web/pages/l/[section].tsx b/packages/web/pages/l/[section].tsx index fc943e05b..782e7e14f 100644 --- a/packages/web/pages/l/[section].tsx +++ b/packages/web/pages/l/[section].tsx @@ -4,10 +4,10 @@ import { NavigationLayout, NavigationSection, } from '../../components/templates/NavigationLayout' -import { HomeContainer } from '../../components/nav-containers/home' +import { HomeContainer } from '../../components/nav-containers/HomeContainer' import { LibraryContainer } from '../../components/templates/library/LibraryContainer' import { useMemo } from 'react' -import { HighlightsContainer } from '../../components/nav-containers/highlights' +import { HighlightsContainer } from '../../components/nav-containers/HighlightsContainer' export default function Home(): JSX.Element { const router = useRouter() diff --git a/packages/web/pages/tools/bulk.tsx b/packages/web/pages/tools/bulk.tsx index 24efa499f..f1142d20b 100644 --- a/packages/web/pages/tools/bulk.tsx +++ b/packages/web/pages/tools/bulk.tsx @@ -202,7 +202,7 @@ export default function BulkPerformer(): JSX.Element {