"""Product workstation timed release Logic CRUD operation.""" from datetime import timedelta from dateutil import parser from oto import response from oto.status import BAD_REQUEST, INTERNAL_ERROR, NOT_FOUND, OK from sentry_sdk import capture_exception from timed_release.config import ( TRIGGER_SFN_WS_UPDATE_METADATA_ARN, WS_SCHEDULER_GROUP, WS_SCHEDULER_ROLE_ARN, WS_SEND_NOTIFICATION_LAMBDA_ARN, ) from timed_release.connectors import scheduler, sql from timed_release.constants import error from timed_release.constants.timed_release import ( EVENT_SCHEDULE_OFFSET, MESSAGE_SAME_TIMED_RELEASE_SET_IN_WORKSTATION, TIMED, UPSERT_SCHEDULE, UTC_DATETIME_FOTMAT, WARNING_EMAIL, WARNING_SCHEDULE_OFFSET, WORKSTATION, ) from timed_release.models import ows_store, scheduled_update, timed_release def _get_combined_timed_release(product_id, session): """ Retrieve and validate the combined timed release date for a product. This function consolidates sales_date_time values from both supported and unsupported store timed release configurations via workstation (WS), ensuring they resolve to a single consistent date-time. Workflow: 1. Validate that no timed release sets exist in OA for supported or unsupported stores. If found, return 400 BAD_REQUEST. 2. Fetch supported store timed release set from WS: - If Ok, consolidate to unique sales_date_time using set - If not found, continue. - If multiple sales_date_time values exist, return 400 BAD_REQUEST. 3. Fetch unsupported store timed release set from WS: - If not found, continue. - Add its sales_date_time to the set. - If multiple unique values exist after merging, return 400 BAD_REQUEST. 4. Return 200 OK with product_id and the single unified sales_date_time (or None if no timed releases exist). Args: product_id (int): Unique identifier of the product. session (Session): Database session object used for queries. Returns: response.Response: - 200 OK: Contains product_id and unified sales_date_time. - 400 BAD_REQUEST: Inconsistent or multiple sales_date_times found. - 404 NOT_FOUND: Timed release entries not found in WS. - Other: Propagates unexpected WS response as-is. """ if timed_release.check_if_supported_stores_timed_release_oa_exists( product_id, session): return response.Response( status=BAD_REQUEST, message=error.ERROR_MESSAGE_SUPPORTED_TIMED_RELEASE_SET_IN_OA ) if scheduled_update.check_if_unsupported_stores_schedule_oa_exists( product_id, session): return response.Response( status=BAD_REQUEST, message=error.ERROR_MESSAGE_UNSUPPORTED_TIMED_RELEASE_SET_IN_OA ) sales_date_time_set = set() supported_tr_ws_response = ( timed_release.get_supported_stores_timed_release_ws( product_id, session ) ) if supported_tr_ws_response.status == OK: timed_releases = supported_tr_ws_response.message['timed_releases'] sales_date_time_set = ( {row['sales_date_time'] for row in timed_releases} ) elif supported_tr_ws_response.status != NOT_FOUND: return supported_tr_ws_response if len(sales_date_time_set) > 1: return response.Response( status=BAD_REQUEST, message=error.ERROR_MESSAGE_MULTIPLE_SALES_DATE_TIME ) unsupported_tr_ws_response = ( scheduled_update.get_unsupported_stores_timed_release_ws( product_id, session ) ) if unsupported_tr_ws_response.status == OK: unsupported_sales_date_time = ( unsupported_tr_ws_response.message['sales_date_time'] ) sales_date_time_set.add(unsupported_sales_date_time) elif unsupported_tr_ws_response.status != NOT_FOUND: return unsupported_tr_ws_response if len(sales_date_time_set) > 1: return response.Response( status=BAD_REQUEST, message=error.ERROR_MESSAGE_DIFFERENT_SALES_DATE_TIME ) if len(sales_date_time_set) == 1: sales_date_time, = sales_date_time_set else: sales_date_time = None return response.Response( status=OK, message={ 'product_id': product_id, 'sales_date_time': sales_date_time } ) @sql.db_session_wrap def upsert_timed_release_and_schedule( product_id, data, identity_id, request_headers, session=None): """Upsert timed release data and scheduled updates for a given product. This function handles both supported and unsupported store updates: - For supported stores: Upserts timed release data in the workstation. - For unsupported stores: Creates or updates scheduled updates to trigger sales at the specified time (with a process trigger 30 minutes earlier). The function ensures that duplicate timed release data is not set if the same sales date-time already exists in the workstation. Additionally, this upserts the event schedule in aws event bridge. Args: product_id (int): The product ID for which timed release and schedule data should be updated. data (dict): Payload containing 'sales_date_time'. identity_id (str): Identity ID of the user performing the operation. request_headers (dict): Headers sent in the API request (used to fetch active stores). session (Session, optional): Database session. Defaults to None. Returns: response.Response: - 200 OK with details of upserted timed releases and schedules on success. Also, if the same timed release is already set in the workstation. - Error response if fetching stores, upserting timed release, or upserting schedule fails. """ sales_date_time_str = data['sales_date_time'] ws_timed_release_response = _get_combined_timed_release( product_id, session) if ( ws_timed_release_response.status == OK and sales_date_time_str == ws_timed_release_response.message['sales_date_time'] ): return response.Response( status=OK, message=MESSAGE_SAME_TIMED_RELEASE_SET_IN_WORKSTATION ) elif ws_timed_release_response.status != OK: return ws_timed_release_response get_stores_response = ows_store.get_stores(request_headers) if get_stores_response.status != OK: return get_stores_response active_stores = get_stores_response.message supported_stores_tr_data = [] unsupported_stores_schedule_data = [] unsupported_store_ids = [] upserted_data = [] user_timezone = data.get('user_timezone') for store in active_stores: if store['supports_timed_release'] is True: supported_stores_tr_data.append({ 'product_id': product_id, 'timing': TIMED, 'store_id': store['id'], 'sales_date_time': sales_date_time_str, 'source': WORKSTATION, 'user_timezone': user_timezone }) else: unsupported_store_ids.append(store['id']) upserted_supported_releases_response = ( timed_release.upsert_supported_stores_timed_release_ws( product_id, supported_stores_tr_data, identity_id, session) ) if upserted_supported_releases_response.status == OK: upserted_data.append(upserted_supported_releases_response.message) else: return upserted_supported_releases_response sales_date_time = parser.isoparse(sales_date_time_str) process_at_datetime = ( sales_date_time - timedelta(minutes=EVENT_SCHEDULE_OFFSET) ) if unsupported_store_ids: unsupported_stores_schedule_data = { 'delivery_store_ids': unsupported_store_ids, 'schedule_datetime': sales_date_time, 'process_at_datetime': process_at_datetime, 'process_at_offset': f'00:{EVENT_SCHEDULE_OFFSET}', 'update': { 'sale_start_date': sales_date_time.date() }, 'user_timezone': user_timezone } upserted_unsupported_schedule_response = ( scheduled_update.upsert_unsupported_stores_schedule_ws( product_id, unsupported_stores_schedule_data, identity_id, session ) ) if upserted_unsupported_schedule_response.status == OK: upserted_data.append( upserted_unsupported_schedule_response.message) else: return upserted_unsupported_schedule_response event_schedule_datetime = process_at_datetime.strftime(UTC_DATETIME_FOTMAT) try: scheduler.upsert_event_schedule( schedule_name=product_id, scheduler_group=WS_SCHEDULER_GROUP, scheduler_role_arn=WS_SCHEDULER_ROLE_ARN, schedule_datetime=event_schedule_datetime, description=f'Metadata update for Product: {product_id}', target_arn=TRIGGER_SFN_WS_UPDATE_METADATA_ARN, payload={ 'productId': product_id, 'eventType': UPSERT_SCHEDULE, 'salesDateTime': sales_date_time_str, 'source': WORKSTATION } ) try: warn_at = ( process_at_datetime - timedelta(hours=WARNING_SCHEDULE_OFFSET) ) warn_at_datetime = warn_at.strftime(UTC_DATETIME_FOTMAT) scheduler.upsert_event_schedule( schedule_name=f'{product_id}-warning', scheduler_group=WS_SCHEDULER_GROUP, scheduler_role_arn=WS_SCHEDULER_ROLE_ARN, description=( f'Pre-metadata update warning for ' f'Product: {product_id}' ), schedule_datetime=warn_at_datetime, target_arn=WS_SEND_NOTIFICATION_LAMBDA_ARN, payload={ 'productId': product_id, 'eventProcessOn': event_schedule_datetime, 'eventType': WARNING_EMAIL, 'salesDateTime': sales_date_time_str, 'source': WORKSTATION } ) except Exception as e: scheduler.delete_event_schedule(product_id, WS_SCHEDULER_GROUP) raise e except Exception: session.rollback() capture_exception() return response.create_error_response( code=error.INTERNAL_ERROR, message=error.ERROR_MESSAGE_UPSERTING_WS_SCHEDULE_FAILED, status=INTERNAL_ERROR ) return response.Response(status=OK, message=upserted_data) def get_product_timed_release_for_ws(product_id): """Get unique product timed_release set via workstation. Check if OA set timed release exists for supported or unsupported stores. If exists for either supported and unsupported stores, return 200 with response data containing is_oa_set_timed_release set to true. Fetch supported store timed release set from WS: If not found, continue. If OK, consolidate to unique sales_date_time using set. If multiple sales_date_time values exist, return 400 BAD_REQUEST. Fetch unsupported store timed release set from WS: If OK, consolidate to unique sales_date_time using set. If multiple sales_date_time values exist, return 400 BAD_REQUEST. If timed release not found for both supported and unsupported stores, return below response. If unique unified sales_date_time, return 200 OK with product_id and the single unified sales_date_time. Args: product_id (int): Unique identifier of the product. Returns: response.Response: - 200 OK: If unified sales_date_time or not found. - 400 BAD_REQUEST: Inconsistent or multiple sales_date_times found. - Other: Propagates unexpected WS response as-is. """ with sql.db_session(read_only=True) as session: if timed_release.check_if_supported_stores_timed_release_oa_exists( product_id, session): return response.Response( status=OK, message={ 'product_id': product_id, 'sales_date_time': None, 'is_oa_set_timed_release': True, 'message': ( error.ERROR_MESSAGE_SUPPORTED_TIMED_RELEASE_SET_IN_OA ) } ) if scheduled_update.check_if_unsupported_stores_schedule_oa_exists( product_id, session): return response.Response( status=OK, message={ 'product_id': product_id, 'sales_date_time': None, 'is_oa_set_timed_release': True, 'message': ( error.ERROR_MESSAGE_UNSUPPORTED_TIMED_RELEASE_SET_IN_OA ) } ) sales_date_time_set = set() user_timezone_set = set() supported_tr_ws_response = ( timed_release.get_supported_stores_timed_release_ws( product_id, session ) ) if supported_tr_ws_response.status == OK: timed_releases = supported_tr_ws_response.message['timed_releases'] sales_date_time_set = ( {row['sales_date_time'] for row in timed_releases} ) user_timezone_set = ( {row['user_timezone'] for row in timed_releases} ) elif supported_tr_ws_response.status != NOT_FOUND: return supported_tr_ws_response if len(sales_date_time_set) > 1: return response.Response( status=BAD_REQUEST, message=error.ERROR_MESSAGE_MULTIPLE_SALES_DATE_TIME ) unsupported_tr_ws_response = ( scheduled_update.get_unsupported_stores_timed_release_ws( product_id, session ) ) if unsupported_tr_ws_response.status == OK: unsupported_sales_date_time = ( unsupported_tr_ws_response.message['sales_date_time'] ) unsupported_user_timezone = ( unsupported_tr_ws_response.message['user_timezone'] ) sales_date_time_set.add(unsupported_sales_date_time) user_timezone_set.add(unsupported_user_timezone) elif unsupported_tr_ws_response.status != NOT_FOUND: return unsupported_tr_ws_response if len(sales_date_time_set) > 1: return response.Response( status=BAD_REQUEST, message=error.ERROR_MESSAGE_DIFFERENT_SALES_DATE_TIME ) if len(user_timezone_set) > 1: return response.Response( status=BAD_REQUEST, message=error.ERROR_MESSAGE_MULTIPLE_USER_TIMEZONE ) if len(sales_date_time_set) == 1: sales_date_time, = sales_date_time_set else: sales_date_time = None if len(user_timezone_set) == 1: user_timezone, = user_timezone_set else: user_timezone = None return response.Response( status=OK, message={ 'product_id': product_id, 'sales_date_time': sales_date_time, 'user_timezone': user_timezone, 'is_oa_set_timed_release': False } ) @sql.db_session_wrap def delete_timed_releases_and_scheduled_updates_ws(product_id, session=None): """Delete workstation set timed_releases and scheduled_updates for product. First attempts to hard delete all timed releases for the product. If timed release deletion fails, return the error. Else proceeds to fetch scheduled_update data in order to use this data when the warning schedule delete fails, since it's a due to hard delete we would need this data to restore the event schedule. Fetch scheduled_update data If fail, return error. If not found, skip scheduled_update data deletion. Else, attempt to hard delete scheduled_update data. If hard delete fails, return error If both timed release and scheduled update not found, then return 404. Upon successful deletion of scheduled_update Delete event schedule from the aws EventBridge for workstation scheduler group. If event schedule delete fails, return error. Else, attempt to delete warning schedule. If the warning schedule deletion fails, it recreates the event schedule and raises the exception. In case of any internal error during deletion, the session is rolled back and a 500 Internal Server Error response is returned. Finally, return 200 in case of successful delete. Args: product_id (int): Product ID for which the schedule should be deleted. orchard_identity_id (str): Orchard Identity ID of the requesting user. session (object): SQLAlchemy session object for DB operations. This is handled by @sql.db_session_wrap decorator. Returns: response.Response: - 200 OK: Atleast one or both deletion succeeded. - 404 Not Found: timed releases and scheduled updates not found. - 500 Internal Server Error: Database or deletion operation fail. """ try: delete_timed_release_ws_resp = timed_release.delete_timed_release_ws( product_id, session) if delete_timed_release_ws_resp.status == INTERNAL_ERROR: return delete_timed_release_ws_resp get_scheduled_update_ws_resp = ( scheduled_update.get_scheduled_update_ws(product_id, session) ) delete_scheduled_update_ws_status = None if get_scheduled_update_ws_resp.status == INTERNAL_ERROR: return get_scheduled_update_ws_resp if get_scheduled_update_ws_resp.status == OK: delete_scheduled_update_ws_resp = ( scheduled_update.delete_scheduled_update_ws( product_id, session) ) delete_scheduled_update_ws_status = ( delete_scheduled_update_ws_resp.status ) if delete_scheduled_update_ws_resp.status == INTERNAL_ERROR: return delete_scheduled_update_ws_resp if ( delete_timed_release_ws_resp.status == NOT_FOUND and ( get_scheduled_update_ws_resp.status == NOT_FOUND or delete_scheduled_update_ws_status == NOT_FOUND ) ): return response.Response(status=NOT_FOUND) if delete_scheduled_update_ws_status == OK: scheduler.delete_event_schedule(product_id, WS_SCHEDULER_GROUP) try: scheduler.delete_event_schedule( f'{product_id}-warning', WS_SCHEDULER_GROUP) except Exception as e: scheduled_update_ws = get_scheduled_update_ws_resp.message scheduler.upsert_event_schedule( schedule_name=product_id, scheduler_group=WS_SCHEDULER_GROUP, scheduler_role_arn=WS_SCHEDULER_ROLE_ARN, schedule_datetime=( scheduled_update_ws['process_at_datetime'] ), description=f'Metadata update for Product: {product_id}', target_arn=TRIGGER_SFN_WS_UPDATE_METADATA_ARN, payload={ 'productId': product_id, 'eventType': UPSERT_SCHEDULE } ) raise e except Exception: session.rollback() capture_exception() return response.create_error_response( code=error.INTERNAL_ERROR, message=error.ERROR_MESSAGE_WS_SCHEDULE_DELETE_FAILED, status=INTERNAL_ERROR ) return response.Response(status=OK)