def deactivate_orphaned_flight(az_flight_id):
    g.log.debug("queuing deactivate_orphaned_flight %d" % az_flight_id)

    amqp.add_item("adzerk_q", json.dumps({
        "action": "deactivate_orphaned_flight",
        "flight": az_flight_id,
    }))

def run(verbose=True, sleep_time = 60, num_items = 1):
    key = "indextank_cursor"
    cursor = g.cache.get(key)
    if cursor is None:
        raise ValueError("%s is not set!" % key)
    cursor = int(cursor)

    while True:
        if verbose:
            print "Looking for %d items with _id < %d" % (num_items, cursor)
        q = Link._query(sort = desc('_id'),
                        limit = num_items)
        q._after(Link._byID(cursor))
        last_date = None
        for item in q:
            cursor = item._id
            last_date = item._date
            amqp.add_item('indextank_changes', item._fullname,
                      message_id = item._fullname,
                      delivery_mode = amqp.DELIVERY_TRANSIENT)
        g.cache.set(key, cursor)

        if verbose:
            if last_date:
                last_date = last_date.strftime("%Y-%m-%d")
            print ("Just enqueued %d items. New cursor=%s (%s). Sleeping %d seconds."
                   % (num_items, cursor, last_date, sleep_time))

        sleep(sleep_time)

def new_comment(comment, inbox_rels):
    author = Account._byID(comment.author_id)
    job = [get_comments(author, 'new', 'all')]
    if comment._deleted:
        job.append(get_all_comments())
        add_queries(job, delete_items = comment)
    else:
        if comment._spam:
            sr = Subreddit._byID(comment.sr_id)
            job.append(get_spam_comments(sr))
        add_queries(job, insert_items = comment)
        amqp.add_item('new_comment', comment._fullname)
        if not g.amqp_host:
            add_comment_tree([comment])

    # note that get_all_comments() is updated by the amqp process
    # r2.lib.db.queries.run_new_comments (to minimise lock contention)

    if inbox_rels:
        for inbox_rel in tup(inbox_rels):
            inbox_owner = inbox_rel._thing1
            if inbox_rel._name == "inbox":
                add_queries([get_inbox_comments(inbox_owner)],
                            insert_items = inbox_rel)
            else:
                add_queries([get_inbox_selfreply(inbox_owner)],
                            insert_items = inbox_rel)
            set_unread(comment, inbox_owner, True)

def add_to_author_query_q(link):
    if g.shard_author_query_queues:
        author_shard = link.author_id % 10
        queue_name = "author_query_%s_q" % author_shard
    else:
        queue_name = "author_query_q"
    amqp.add_item(queue_name, link._fullname)

def cast_vote(user, thing, direction, **data):
    """Register a vote and queue it for processing."""
    update_vote_lookups(user, thing, direction)

    vote_data = {
        "user_id": user._id,
        "thing_fullname": thing._fullname,
        "direction": direction,
        "date": int(epoch_timestamp(datetime.now(g.tz))),
    }

    data['ip'] = getattr(request, "ip", None)
    if data['ip'] is not None:
        data['org'] = organization_by_ips(data['ip'])
    vote_data['data'] = data

    hooks.get_hook("vote.get_vote_data").call(
        data=vote_data["data"],
        user=user,
        thing=thing,
        request=request,
        context=c,
    )

    # The vote event will actually be sent from an async queue processor, so
    # we need to pull out the context data at this point
    if not g.running_as_script:
        vote_data["event_data"] = {
            "context": Event.get_context_data(request, c),
            "sensitive": Event.get_sensitive_context_data(request, c),
        }

    amqp.add_item(thing.vote_queue_name, json.dumps(vote_data))

def changed(things):
    """Indicate to solrsearch that a given item should be updated"""
    things = tup(things)
    for thing in things:
        amqp.add_item('searchchanges_q', thing._fullname,
                      message_id = thing._fullname,
                      delivery_mode = amqp.DELIVERY_TRANSIENT)

def new_comment(comment, inbox_rels):
    author = Account._byID(comment.author_id)
    job = [get_comments(author, "new", "all")]
    if comment._deleted:
        job.append(get_all_comments())
        add_queries(job, delete_items=comment)
    else:
        # if comment._spam:
        #    sr = Subreddit._byID(comment.sr_id)
        #    job.append(get_spam_comments(sr))
        add_queries(job, insert_items=comment)
        amqp.add_item("new_comment", comment._fullname)
        if not g.amqp_host:
            l = Link._byID(comment.link_id, data=True)
            add_comment_tree(comment, l)

    # note that get_all_comments() is updated by the amqp process
    # r2.lib.db.queries.run_new_comments

    if inbox_rels:
        for inbox_rel in tup(inbox_rels):
            inbox_owner = inbox_rel._thing1
            if inbox_rel._name == "inbox":
                add_queries([get_inbox_comments(inbox_owner)], insert_items=inbox_rel)
            else:
                add_queries([get_inbox_selfreply(inbox_owner)], insert_items=inbox_rel)
            set_unread(comment, inbox_owner, True)

def log_text(classification, text=None, level="info"):
    """Send some log text to log_q for appearance in the streamlog.

    This is deprecated. All logging should be done through python's stdlib
    logging library.

    """

    from r2.lib import amqp
    from r2.lib.filters import _force_utf8

    if text is None:
        text = classification

    if level not in ("debug", "info", "warning", "error"):
        print "What kind of loglevel is %s supposed to be?" % level
        level = "error"

    d = _default_dict()
    d["type"] = "text"
    d["level"] = level
    d["text"] = _force_utf8(text)
    d["classification"] = classification

    amqp.add_item(QUEUE_NAME, cPickle.dumps(d))

def add_to_subreddit_query_q(link):
    if g.shard_subreddit_query_queues:
        subreddit_shard = link.sr_id % 10
        queue_name = "subreddit_query_%s_q" % subreddit_shard
    else:
        queue_name = "subreddit_query_q"
    amqp.add_item(queue_name, link._fullname)

def push(action, payload):
    g.log.debug("%s: queuing action \"%s\"" % (DFP_QUEUE, action))
    message = json.dumps({
        "action": action,
        "payload": payload,
    })
    amqp.add_item(DFP_QUEUE, message)

def queue_vote(user, thing, dir, ip, organic = False,
               cheater = False, store = True):
    # set the vote in memcached so the UI gets updated immediately
    key = prequeued_vote_key(user, thing)
    g.cache.set(key, '1' if dir is True else '0' if dir is None else '-1')
    # queue the vote to be stored unless told not to
    if store:
        if g.amqp_host:
            if isinstance(thing, Link):
                if thing._id36 in g.live_config["fastlane_links"]:
                    qname = vote_fastlane_q
                else:
                    qname = vote_link_q

            elif isinstance(thing, Comment):
                if utils.to36(thing.link_id) in g.live_config["fastlane_links"]:
                    qname = vote_fastlane_q
                else:
                    qname = vote_comment_q
            else:
                log.warning("%s tried to vote on %r. that's not a link or comment!",
                            user, thing)
                return

            amqp.add_item(qname,
                          pickle.dumps((user._id, thing._fullname,
                                        dir, ip, organic, cheater)))
        else:
            handle_vote(user, thing, dir, ip, organic)

def deactivate_overdelivered(link, campaign):
    g.log.debug('queuing deactivate_overdelivered %s %s' % (link, campaign))
    msg = json.dumps({
        'action': 'deactivate_overdelivered',
        'link': link._fullname,
        'campaign': campaign._fullname,
    })
    amqp.add_item('adzerk_q', msg)

def Run(offset=0):
    """reddit-job-update_promos: Intended to be run hourly to pull in
    scheduled changes to ads
    
    """
    charge_pending(offset=offset + 1)
    charge_pending(offset=offset)
    amqp.add_item(UPDATE_QUEUE, json.dumps(QUEUE_ALL), delivery_mode=amqp.DELIVERY_TRANSIENT)

def queue_modmail_email(message):
    amqp.add_item(
        "modmail_email_q",
        json.dumps({
            "event": "new_message",
            "message_id36": message._id36,
        }),
    )

def update_adzerk(link, campaign=None):
    g.log.debug('queuing update_adzerk %s %s' % (link, campaign))
    msg = json.dumps({
        'action': 'update_adzerk',
        'link': link._fullname,
        'campaign': campaign._fullname if campaign else None,
    })
    amqp.add_item('adzerk_q', msg)

    def update_search_index(self, boost_only=False):
        msg = {'fullname': self._fullname}
        if boost_only:
            msg['boost_only'] = True

        amqp.add_item('search_changes', pickle.dumps(msg),
                      message_id=self._fullname,
                      delivery_mode=amqp.DELIVERY_TRANSIENT)

def send_broadcast(namespace, message):
    """Broadcast an object to all WebSocket listeners in a namespace.

    The message will be encoded as a JSON object before being sent to the
    client.

    """
    amqp.add_item(routing_key=namespace, body=json.dumps(message),
                  exchange=_WEBSOCKET_EXCHANGE)

def changed(things, boost_only=False):
    """Indicate to search that a given item should be updated in the index"""
    for thing in tup(things):
        msg = {'fullname': thing._fullname}
        if boost_only:
            msg['boost_only'] = True

        amqp.add_item('search_changes', pickle.dumps(msg),
                      message_id = thing._fullname,
                      delivery_mode = amqp.DELIVERY_TRANSIENT)

def queue_vote(user, thing, dir, ip, organic=False, cheater=False, store=True):
    # set the vote in memcached so the UI gets updated immediately
    key = prequeued_vote_key(user, thing)
    g.cache.set(key, "1" if dir is True else "0" if dir is None else "-1")
    # queue the vote to be stored unless told not to
    if store:
        if g.amqp_host:
            amqp.add_item("register_vote_q", pickle.dumps((user._id, thing._fullname, dir, ip, organic, cheater)))
        else:
            handle_vote(user, thing, dir, ip, organic)

def _generate_link_reports(items):
    links = items["links"]
    campaigns = items["campaigns"]

    g.log.info("queuing report for link %s" % ",".join(l._fullname for l in links))
    amqp.add_item("adzerk_reporting_q", json.dumps({
        "action": "generate_daily_link_reports",
        "link_ids": [l._id for l in links],
        "campaign_ids": [c._id for c in campaigns],
    }))