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

"""Notification module.

Provides Manager.send_async, which sends notifications; and Manager.query, which
queries the current status of notifications.

Notifications are transported by email. Every message you send consumes email
quota. A message is a single payload delivered to a single user. We do not
provide the entire interface email does (no CC, BCC, attachments, or HTML
bodies). Note that messages are not sent when you call Manager.send_async(), but
instead enqueued and sent later -- usually within a minute.

This module has several advantages over using App Engine's mail.send_mail()
directly.

First, we queue and retry sending messages. This happens on two levels: first,
send_async() adds items to a task queue, which retries if there are transient
failures (like the datastore being slow, or you enqueueing more messages than
App Engine's mail service can send in a minute). Second, we provide a cron that
retries message delivery for several days, so if you exhaust your daily mail
quota today we'll try again tomorrow.

The second major advantage is that we keep a record of messages sent, so you can
do analytics on them. We provide a base set of dashboards in the admin UI
showing both overall and recent notification state.

For users who are sending mail occasionally, this module smoothes away some of
the gotchas of App Engine's mail service. However, App Engine is not optimized
to be a bulk mail delivery service, so if you need to send amounts of mail in
excess of App Engine's max daily quota (1.7M messages) or minute-by-minute quota
(5k messages), you should consider using a third-party mail delivery service.

We provide a second module that allows your users to opt out of receiving email.
We strongly encourage use of that module so you don't spam people. See
modules/unsubscribe/unsubscribe.py. The general pattern for using these modules
is:

  from modules.notifications import notifications
  from modules.unsubscribe import unsubscribe

  from google.appengine.api import users

  user = users.get_current_user()

  if user and not unsubscribe.has_unsubscribed(user.email):
    notifications.Manager.send_async(
        user.email, 'sender@example.com', 'intent', 'subject', 'body'
    )
"""

__author__ = [
  'johncox@google.com (John Cox)'
]

import datetime
import logging

from models import counters
from models import custom_modules
from models import entities
from models import services
from models import transforms
from models import utils
from modules import dashboard

from google.appengine.api import mail
from google.appengine.api import mail_errors
from google.appengine.api import taskqueue
from google.appengine.datastore import datastore_rpc
from google.appengine.ext import db
from google.appengine.ext import deferred


_LOG = logging.getLogger('modules.notifications.notifications')
logging.basicConfig()


_APP_ENGINE_MAIL_FATAL_ERRORS = frozenset([
    mail_errors.BadRequestError, mail_errors.InvalidSenderError,
])
_ENQUEUED_BUFFER_MULTIPLIER = 1.5
_KEY_DELIMITER = ':'
_MAX_ENQUEUED_HOURS = 3
_MAX_RETRY_DAYS = 3
# Number of times past which recoverable failure of send_mail() calls becomes
# hard failure. Used as a brake on runaway queues. Should be larger than the
# expected cap on the number of retries imposed by taskqueue.
_RECOVERABLE_FAILURE_CAP = 20
_SECONDS_PER_HOUR = 60 * 60
_SECONDS_PER_DAY = 24 * _SECONDS_PER_HOUR
_USECS_PER_SECOND = 10 ** 6

COUNTER_RETENTION_POLICY_RUN = counters.PerfCounter(
    'gcb-notifications-retention-policy-run',
    'number of times a retention policy was run'
)
COUNTER_SEND_ASYNC_FAILED_BAD_ARGUMENTS = counters.PerfCounter(
    'gcb-notifications-send-async-failed-bad-arguments',
    'number of times send_async failed because arguments were bad'
)
COUNTER_SEND_ASYNC_FAILED_DATASTORE_ERROR = counters.PerfCounter(
    'gcb-notifications-send-async-failed-datastore-error',
    'number of times send_async failed because of datastore error'
)
COUNTER_SEND_ASYNC_START = counters.PerfCounter(
    'gcb-notifications-send-async-called',
    'number of times send_async has been called'
)
COUNTER_SEND_ASYNC_SUCCESS = counters.PerfCounter(
    'gcb-notifications-send-async-success',
    'number of times send_async succeeded'
)
COUNTER_SEND_MAIL_TASK_FAILED = counters.PerfCounter(
    'gcb-notifications-send-mail-task-failed',
    'number of times the send mail task failed, but could be retried'
)
COUNTER_SEND_MAIL_TASK_FAILED_PERMANENTLY = counters.PerfCounter(
    'gcb-notifications-send-mail-task-failed-permanently',
    'number of times the send mail task failed permanently'
)
COUNTER_SEND_MAIL_TASK_FAILURE_CAP_EXCEEDED = counters.PerfCounter(
    'gcb-notifications-send-mail-task-recoverable-failure-cap-exceeded',
    'number of times the recoverable failure cap was exceeded'
)
COUNTER_SEND_MAIL_TASK_RECORD_FAILURE_CALLED = counters.PerfCounter(
    'gcb-notifications-send-mail-task-record-failure-called',
    'number of times _record_failure was called in the send mail task'
)
COUNTER_SEND_MAIL_TASK_RECORD_FAILURE_FAILED = counters.PerfCounter(
    'gcb-notifications-send-mail-task-record-failure-failed',
    'number of times _record_failure failed in the send mail task'
)
COUNTER_SEND_MAIL_TASK_RECORD_FAILURE_SUCCESS = counters.PerfCounter(
    'gcb-notifications-send-mail-task-record-failure-success',
    'number of times _record_failure succeeded in the send mail task'
)
COUNTER_SEND_MAIL_TASK_SENT = counters.PerfCounter(
    'gcb-notifications-send-mail-task-sent',
    'number of times the send mail task called send_mail successfully'
)
COUNTER_SEND_MAIL_TASK_SKIPPED = counters.PerfCounter(
    'gcb-notifications-send-mail-task-skipped',
    'number of times send mail task skipped sending mail'
)
COUNTER_SEND_MAIL_TASK_STARTED = counters.PerfCounter(
    'gcb-notifications-send-mail-task-started',
    'number of times the send mail task was dequeued and started')
COUNTER_SEND_MAIL_TASK_SUCCESS = counters.PerfCounter(
    'gcb-notifications-send-mail-task-success',
    'number of times send mail task completed successfully'
)


# TODO(johncox): remove suppression once stubs are implemented.
# pylint: disable=unused-argument


def _dt_to_epoch_usec(dt):
  """Converts datetime (assumed UTC) to epoch microseconds."""
  return int((_USECS_PER_SECOND) * (
      dt - datetime.datetime.utcfromtimestamp(0)).total_seconds())


def _epoch_usec_to_dt(usec):
  """Converts microseconds since epoch int to datetime (UTC, no tzinfo)."""
  return (
      datetime.datetime.utcfromtimestamp(0) +
      datetime.timedelta(microseconds=usec)
  )


class Error(Exception):
  """Base error class."""


class NotificationTooOldError(Error):
  """Recorded on a notification by cron when it's too old to re-enqueue."""


class RetentionPolicy(object):
  """Retention policy for notification data.

  Notification data is spread between the Notification and Payload objects (of
  which see below). Two parts of this data may be large:
  Notification.audit_trail, and Payload.body.

  We allow clients to specify a retention policy when calling
  Manager.send_async(). This retention policy is a bundle of logic applied after
  we know a notification has been sent. How and when the retention policy is run
  is up to the implementation; we make no guarantees except that once the
  notification is sent we will attempt run() at least once, and if it mutates
  its input we will attempt to apply those mutations at least once.

  Practically, it can be used to prevent retention of data in the datastore that
  is of no use to the client, even for audit purposes.

  Note that 'retention' here has nothing to do with broader user data privacy
  and retention concerns -- this is purely about responsible resource usage.
  """

  # String. Name used to identify the retention policy (in the datastore, for)
  # example.
  NAME = None

  @classmethod
  def run(cls, notification, payload):
    """Runs the policy, transforming notification and payload in place.

    run does not apply mutations to the backing datastore entities; it merely
    returns versions of those entities that we will later attempt to persist.
    Your transforms must not touch protected fields on notification or payload;
    those are used by the subsystem, and changing them can violate constraints
    and cause unpredictable behavior and data corruption.

    Args:
      notification: Notification. The notification to process.
      payload: Payload. The payload to process.
    """
    pass


class RetainAll(RetentionPolicy):
  """Policy that retains all data."""

  NAME = 'all'


class RetainAuditTrail(RetentionPolicy):
  """Policy that blanks Payload.body but not Notification.audit_trail."""

  NAME = 'audit_trail'

  @classmethod
  def run(cls, unused_notification, payload):
    payload.body = None


# Dict of string -> RetentionPolicy where key is the policy's NAME. All
# available retention policies.
_RETENTION_POLICIES = {
    RetainAll.NAME: RetainAll,
    RetainAuditTrail.NAME: RetainAuditTrail,
}


class Status(object):
  """DTO for email status."""

  FAILED = 'failed'
  PENDING = 'pending'
  SUCCEEDED = 'succeeded'
  _STATES = frozenset((FAILED, PENDING, SUCCEEDED))

  def __init__(self, to, sender, intent, enqueue_date, state):
    assert state in self._STATES

    self.enqueue_date = enqueue_date
    self.intent = intent
    self.sender = sender
    self.state = state
    self.to = to

  @classmethod
  def from_notification(cls, notification):
    state = cls.PENDING

    # Treating as module-protected. pylint: disable=protected-access
    if notification._fail_date:
      state = cls.FAILED
    elif notification._done_date:
      state = cls.SUCCEEDED

    return cls(
        notification.to, notification.sender, notification.intent,
        notification.enqueue_date, state
    )

  def __eq__(self, other):
    return (
        self.enqueue_date == other.enqueue_date and
        self.intent == other.intent and
        self.sender == other.sender and
        self.state == other.state and
        self.to == other.to
    )

  def __str__(self):
    return (
        'Status - to: %(to)s, from: %(sender)s, intent: %(intent)s, '
        'enqueued: %(enqueue_date)s, state: %(state)s' % {
            'enqueue_date': self.enqueue_date,
            'intent': self.intent,
            'sender': self.sender,
            'state': self.state,
            'to': self.to,
    })


def _accumulate_statuses(notification, results):
  for_user = results.get(notification.to, [])
  for_user.append(Status.from_notification(notification))
  results[notification.to] = for_user


class Manager(object):
  """Manages state and operation of the notifications subsystem."""

  # Treating access as module-protected. pylint: disable=protected-access

  @classmethod
  def query(cls, to, intent):
    """Gets the Status of notifications queued previously via send_async().

    Serially performs one datastore query per user in the to list.

    Args:
      to: list of string. The recipients of the notification.
      intent: string. Short string identifier of the intent of the notification
          (for example, 'invitation' or 'reminder').

    Returns:
      Dict of to string -> [Status, sorted by descending enqueue date].
    """
    results = {}

    for address in to:
      mapper = utils.QueryMapper(cls._get_query_query(address, intent))
      mapper.run(_accumulate_statuses, results)

    return results

  @classmethod
  def send_async(
      cls, to, sender, intent, body, subject, audit_trail=None,
      retention_policy=None):
    """Asyncronously sends a notification via email.

    Args:
      to: string. Recipient email address. Must have a valid form, but we cannot
          know that the address can actually be delivered to.
      sender: string. Email address of the sender of the notification. Must be a
          valid sender for the App Engine deployment at the time the deferred
          send_mail() call actually executes (meaning it cannot be the email
          address of the user currently in session, because the user will not be
          in session at call time). See
          https://developers.google.com/appengine/docs/python/mail/emailmessagefields.
      intent: string. Short string identifier of the intent of the notification
          (for example, 'invitation' or 'reminder'). Each kind of notification
          you are sending should have its own intent. Used when creating keys in
          the index; values that cause the resulting key to be >500B will fail.
          May not contain a colon.
      body: string. The data payload of the notification. Must fit in a
          datastore entity.
      subject: string. Subject line for the notification.
      audit_trail: JSON-serializable object. An optional audit trail that, when
          used with the default retention policy, will be retained even after
          the body is scrubbed from the datastore.
      retention_policy: RetentionPolicy. The retention policy to use for data
          after a Notification has been sent. By default, we retain the
          audit_trail but not the body.

    Returns:
      (notification_key, payload_key). A 2-tuple of datastore keys for the
      created notification and payload.

    Raises:
      Exception: if values delegated to model initializers are invalid.
      ValueError: if to or sender are malformed according to App Engine (note
          that well-formed values do not guarantee success).
    """
    COUNTER_SEND_ASYNC_START.inc()
    enqueue_date = datetime.datetime.utcnow()
    retention_policy = (
        retention_policy if retention_policy else RetainAuditTrail)

    for email in (to, sender):
      if not mail.is_email_valid(email):
        COUNTER_SEND_ASYNC_FAILED_BAD_ARGUMENTS.inc()
        raise ValueError('Malformed email address: "%s"' % email)

    if retention_policy.NAME not in _RETENTION_POLICIES:
      COUNTER_SEND_ASYNC_FAILED_BAD_ARGUMENTS.inc()
      raise ValueError('Invalid retention policy: ' + str(retention_policy))

    try:
      notification, payload = cls._make_unsaved_models(
          audit_trail, body, enqueue_date, intent, retention_policy.NAME,
          sender, subject, to,
      )
    except Exception, e:
      COUNTER_SEND_ASYNC_FAILED_BAD_ARGUMENTS.inc()
      raise e

    cls._mark_enqueued(notification, enqueue_date)

    try:
      notification_key, payload_key = cls._save_notification_and_payload(
          notification, payload,
      )
    except Exception, e:
      COUNTER_SEND_ASYNC_FAILED_DATASTORE_ERROR.inc()
      raise e

    deferred.defer(
        cls._transactional_send_mail_task, notification_key, payload_key,
        _retry_options=cls._get_retry_options())
    COUNTER_SEND_ASYNC_SUCCESS.inc()

    return notification_key, payload_key

  @classmethod
  def _make_unsaved_models(
      cls, audit_trail, body, enqueue_date, intent, retention_policy, sender,
      subject, to):
    notification = Notification(
        audit_trail=audit_trail, enqueue_date=enqueue_date, intent=intent,
        _retention_policy=retention_policy, sender=sender, subject=subject,
        to=to,
    )
    payload = Payload(
      body=body, enqueue_date=enqueue_date, intent=intent, to=to,
      _retention_policy=retention_policy,
    )

    return notification, payload

  @classmethod
  @db.transactional(xg=True)
  def _save_notification_and_payload(cls, notification, payload):
    return db.put([notification, payload])

  @classmethod
  def _send_mail_task(
      cls, notification_key, payload_key, test_send_mail_fn=None):
    exception = None
    failed_permanently = False
    now = datetime.datetime.utcnow()
    notification, payload = db.get([notification_key, payload_key])
    send_mail_fn = test_send_mail_fn if test_send_mail_fn else mail.send_mail
    sent = False

    COUNTER_SEND_MAIL_TASK_STARTED.inc()

    if not notification:
      COUNTER_SEND_MAIL_TASK_FAILED_PERMANENTLY.inc()
      raise deferred.PermanentTaskFailure(
          'Notification missing: ' + str(notification_key)
      )

    if not payload:
      COUNTER_SEND_MAIL_TASK_FAILED_PERMANENTLY.inc()
      raise deferred.PermanentTaskFailure(
          'Payload missing: ' + str(payload_key)
      )

    policy = _RETENTION_POLICIES.get(notification._retention_policy)
    if not policy:
      COUNTER_SEND_MAIL_TASK_FAILED_PERMANENTLY.inc()
      raise deferred.PermanentTaskFailure(
          'Unknown retention policy: ' + notification._retention_policy
      )

    if (cls._done(notification) or cls._failed(notification) or
        cls._sent(notification)):
      COUNTER_SEND_MAIL_TASK_SKIPPED.inc()
      COUNTER_SEND_MAIL_TASK_SUCCESS.inc()
      return

    if notification._recoverable_failure_count > _RECOVERABLE_FAILURE_CAP:
      message = (
          'Recoverable failure cap (%s) exceeded for notification with '
          'key %s'
      ) % (_RECOVERABLE_FAILURE_CAP, str(notification.key()))
      _LOG.error(message)
      permanent_failure = deferred.PermanentTaskFailure(message)

      try:
        COUNTER_SEND_MAIL_TASK_RECORD_FAILURE_CALLED.inc()
        cls._record_failure(
            notification, payload, permanent_failure, dt=now, permanent=True,
            policy=policy
        )
        COUNTER_SEND_MAIL_TASK_RECORD_FAILURE_SUCCESS.inc()
      # Must be vague. pylint: disable=broad-except
      except Exception, e:
        _LOG.error(
            cls._get_record_failure_error_message(notification, payload, e)
        )
        COUNTER_SEND_MAIL_TASK_RECORD_FAILURE_FAILED.inc()

      COUNTER_SEND_MAIL_TASK_FAILED_PERMANENTLY.inc()
      COUNTER_SEND_MAIL_TASK_FAILURE_CAP_EXCEEDED.inc()

      raise permanent_failure

    try:
      send_mail_fn(
          notification.sender, notification.to, notification.subject,
          payload.body
      )
      sent = True
    # Must be vague. pylint: disable=broad-except
    except Exception, exception:
      failed_permanently = cls._is_send_mail_error_permanent(exception)

      if not failed_permanently:

        try:
          COUNTER_SEND_MAIL_TASK_RECORD_FAILURE_CALLED.inc()
          cls._record_failure(notification, payload, exception)
          COUNTER_SEND_MAIL_TASK_RECORD_FAILURE_SUCCESS.inc()
        # Must be vague. pylint: disable=broad-except
        except Exception, e:
          _LOG.error(
              cls._get_record_failure_error_message(
                  notification, payload, exception
              )
          )
          COUNTER_SEND_MAIL_TASK_RECORD_FAILURE_FAILED.inc()

        _LOG.error(
            ('Recoverable error encountered when processing notification task; '
             'will retry. Error was: ' + str(exception))
        )
        COUNTER_SEND_MAIL_TASK_FAILED.inc()

        # Set by except: clause above. pylint: disable=raising-bad-type
        raise exception

    if sent:
      cls._mark_sent(notification, now)

    if failed_permanently:
      cls._mark_failed(notification, now, exception, permanent=True)

    if sent or failed_permanently:
      policy.run(notification, payload)
      cls._mark_done(notification, now)

    db.put([notification, payload])

    COUNTER_RETENTION_POLICY_RUN.inc()

    if sent:
      COUNTER_SEND_MAIL_TASK_SENT.inc()
    elif failed_permanently:
      COUNTER_SEND_MAIL_TASK_FAILED_PERMANENTLY.inc()

    COUNTER_SEND_MAIL_TASK_SUCCESS.inc()

  @classmethod
  @db.transactional(
      propagation=datastore_rpc.TransactionOptions.INDEPENDENT, xg=True)
  def _record_failure(
      cls, notification, payload, exception, dt=None, permanent=False,
      policy=None):
    """Marks failure data on entities in an external transaction.

    IMPORTANT: because we're using datastore_rpc.TransactionOptions.INDEPENDENT,
    mutations on notification and payload here are *not* transactionally
    consistent in the caller. Consequently, callers must not read or mutate them
    after calling this method.

    The upside is that this allows us to record failure data on entities inside
    a transaction, and that transaction can throw without rolling back these
    mutations.

    Args:
      notification: Notification. The notification to mutate.
      payload: Payload. The payload to mutate.
      exception: Exception. The exception that prompted the mutation.
      dt: datetime. notification_fail_time and notification._done_time to record
          if permanent is True.
      permanent: boolean. If True, the notification will be marked done and
          the retention policy will be run.
      policy: RetentionPolicy. The retention policy to apply if permanent was
          True.

    Returns:
      (notification_key, payload_key) 2-tuple.
    """
    notification._recoverable_failure_count += 1
    cls._mark_failed(notification, dt, exception, permanent=permanent)

    if permanent:
      assert dt and policy

      cls._mark_done(notification, dt)
      policy.run(notification, payload)
      COUNTER_RETENTION_POLICY_RUN.inc()

    return db.put([notification, payload])

  @classmethod
  def _get_record_failure_error_message(cls, notification, payload, exception):
    return (
        'Unable to record failure for notification with key %s and payload '
        'with key %s; encountered %s error with text: "%s"') % (
            str(notification.key()), str(payload.key()),
            exception.__class__.__name__, str(exception))

  @classmethod
  def _transactional_send_mail_task(cls, notification_key, payload_key):